o cil @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@s,eZdZUeed<eeed<ejed<dS)_SingleRequestself_argflattened_argsfutureN)__name__ __module__ __qualname__r__annotations__rasyncioFuturer)r)F/home/ubuntu/.local/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+2s r+c@sneZdZUeeed<edeefddZedeefddZ edeefddZ ede fd d Z d S) _RuntimeSummaryStatistics start_timesreturncC|jrt|jSdSN)r/minselfr)r)r*min_start_time<z(_RuntimeSummaryStatistics.min_start_timecCs|jr t|jt|jSdSr2)r/sumlenr4r)r)r*mean_start_time@sz)_RuntimeSummaryStatistics.mean_start_timecCr1r2)r/maxr4r)r)r*max_start_timeFr7z(_RuntimeSummaryStatistics.max_start_timecCs t|jSr2)r9r/r4r)r)r* num_requestsJs z&_RuntimeSummaryStatistics.num_requestsN) r#r$r%rfloatr&propertyrr6r:r<intr=r)r)r)r*r.8s r.list_of_flattened_argsr0cs|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))r9).0argsr)r)r* Vz%_batch_args_kwargs..z=All batch requests should have the same number of parameters.rcsg|]}|qSr)r)rBitemidxr)r* brEz&_batch_args_kwargs..)r9poprangeappendr)rA arg_lengths arg_lengthbatched_flattened_argsr)rJr*_batch_args_kwargsOs  rSc @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_funcr0cCsvt|_||_||_||_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'QueuequeuerUrVrW Semaphore semaphoreEventrequests_available_eventsettaskscurr_iteration_start_times_handle_batch_taskr_loop create_task_process_batches4_warn_if_max_batch_size_exceeds_max_ongoing_requestsr5rUrVrWrXr)r)r*__init__is    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)rget_replica_context_deployment_configmax_ongoing_requestsrUrWloggerwarning)r5rkr)r)r*rfs  z@_BatchQueue._warn_if_max_batch_size_exceeds_max_ongoing_requestsnew_max_batch_sizecCs||_|dSzUpdates queue's max_batch_size.N)rUrfr5rnr)r)r*set_max_batch_sizes z_BatchQueue.set_max_batch_sizerequestcCs|j||jdSr2)rZ put_nowaitr^r_)r5rrr)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)rOrZgetrUrVtimer;r'wait_forr^wait TimeoutErrorr9empty get_nowaitclear)r5batchrUrVbatch_start_timeremaining_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 .)r9r)r5rrr)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)rr9rrNrOUSER_CODE_STREAMING_SENTINELS_set_exception_if_not_doneStopAsyncIterationr create_future_set_result_if_not_doner+popleft Exception) r5rrrFINISHED_TOKENfuturesrrKr,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)r5rrrrr,r"rr)r)r*_assign_func_results"s    z _BatchQueue._assign_func_resultsfunccsh|js2|IdH}|||}t|}|j|t|j |<| |j |jrdSdS)z6Loops infinitely and processes queued request batches.N) rc is_closedr_process_batchr'rdr`addrvraadd_done_callback_handle_completed_task)r5rr}promisetaskr)r)r*re3s     z_BatchQueue._process_batchesr}c sp|j4IdHdd|D}t|dkr" WdIdHdSdd|D}zJ|dj}tdd|D\}}|durJ||g|Ri|}n||i|}t|rd|}|||t|IdHn|} || |t|IdHWn!ty} zt d|D]} t | | qWYd} ~ nd} ~ wwWdIdHdSWdIdHdS1IdHswYdS)zProcesses queued request batch.NcSsg|] }|js|qSr))r" cancelled)rBreqr)r)r*rLDsz._BatchQueue._process_batch..rcSg|]}|jqSr))r"rHr)r)r*rLHcSrr))r!rHr)r)r*rLPrz0_process_batch ran into an unexpected exception.) r\r9r rSrrrrrl exceptionr) r5rr}rr rCkwargsfunc_future_or_generatorrrrr"r)r)r*r=sD        &.z_BatchQueue._process_batchrcCs&|j||j|=||dSr2)r`removera_log_if_exceptionr)r5rr)r)r*ris z"_BatchQueue._handle_completed_taskexception_maybecCs4|durt|tjrtddStddSdS)NzTask was cancelledzTask failed unexpectedly) isinstancer'CancelledErrorrldebugr)rr)r)r*rns  z_BatchQueue._log_if_exceptioncCs&|jdus ts dS|jdSr2)rbr is_runningcancelr4r)r)r*__del__vs z_BatchQueue.__del__r2)r#r$r%r@r>rr rhrfrqrrr'r(rtrrr rrrrrrerTaskr staticmethod BaseExceptionrrr)r)r)r*rThs\ -5   3  , rTc @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. rFNrUrVrWrXcCs"d|_||_||_||_||_dSr2)_queuerUrVrWrXrgr)r)r*rhs  z_LazyBatchQueueWrapper.__init__r0cCs(|jdurt|j|j|j|j|_|jS)zXReturns _BatchQueue. Initializes queue when called for the first time. N)rrTrUrVrWrXr4r)r)r*rZs z_LazyBatchQueueWrapper.queuerncCs$||_|jdur|j|dSdSro)rUrrqrpr)r)r*rqs z)_LazyBatchQueueWrapper.set_max_batch_sizenew_batch_wait_timeout_scCs ||_|jdur||j_dSdSr2)rVr)r5rr)r)r*set_batch_wait_timeout_ss  z/_LazyBatchQueueWrapper.set_batch_wait_timeout_scC|jSr2rUr4r)r)r*get_max_batch_sizez)_LazyBatchQueueWrapper.get_max_batch_sizecCrr2rVr4r)r)r*get_batch_wait_timeout_srz/_LazyBatchQueueWrapper.get_batch_wait_timeout_scCstt|jjS)z;Gets summary statistics of current iteration's start times.)r.listrZravaluesr4r)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. rbF)hasattrrZrbdoner4r)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. rb)fileN)rrZioStringIOrb print_stackgetvalue)r5 str_bufferr)r)r*_get_handling_task_stacks  z/_LazyBatchQueueWrapper._get_handling_task_stack)rrrFN)r#r$r%__doc__r@r>rr rhr?rTrZrqrrrr.rboolrstrrr)r)r)r*rs0    rcCsNt|tst|tr|rt|}ntd||dkr%td|dS)Nz)max_batch_size must be integer >= 1, got rFz,max_batch_size must be an integer >= 1, got )rr@r> 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>r@rrrr)r)r*_validate_batch_wait_timeout_ssrrWcCs$t|tr |dkrtd|dS)NrFz4max_concurrent_batches must be an integer >= 1, got )rr@r)rWr)r)r* _validate_max_concurrent_batchess rSelfTypeT) contravariantTRc@*eZdZdedeedeefddZdS)_SyncBatchingMethodself__SyncBatchingMethod__batchr0cCdSr2r))r5rrr)r)r*__call__sz_SyncBatchingMethod.__call__Nr#r$r%rrrrrr)r)r)r*r"rc@r)_AsyncBatchingMethodr_AsyncBatchingMethod__batchr0csdSr2r))r5rrr)r)r*rsz_AsyncBatchingMethod.__call__Nrr)r)r)r*rrr _sync_funccCrr2r))rr)r)r*r} r} _async_funccCrr2r))rr)r)r*r} _sync_methcCrr2r))rr)r)r*r}r _async_methcCrr2r))rr)r)r*r}rrrrFrUrV__BatchDecoratorcCrr2r))rrUrVrWr)r)r*r}%sc @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 argumentsrr0cCrr2r))r5rr)r)r*r3rz_BatchDecorator.__call__rcCrr2r))r5rr)r)r*r7rrcCrr2r))r5rr)r)r*r=rrcCrr2r))r5rr)r)r*rCrN)r#r$r%rrr rrrrr rrrrr)r)r)r*r0s,0    stable) 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_futurer0cSs<|} z|IdH}|j}|jVWn tyYdSwq)z1Generator that handles generator batch functions.TN)r-r,r)rr"async_responser)r)r*batch_handler_generators   z@batch.._batch_decorator..batch_handler_generatorcsTtt||}t|}|dur|dd}j}t}|t||||S)NrG)rrrrZrrrtr)rCrr!r5 batch_queuer")rlazy_batch_queue_wrapperr)r*enqueue_requests   z8batch.._batch_decorator..enqueue_requestcs||}|Sr2r))rCrr)rrr)r*generator_batch_wrappers z@batch.._batch_decorator..generator_batch_wrappercs||IdHSr2r))rCr)rr)r* batch_wrappersz6batch.._batch_decorator..batch_wrapper)rr'r(rrrr_get_max_batch_sizer_get_batch_wait_timeout_srqrrrr)rrrwrapperrVrUrW)rrrrr*_batch_decorators@ zbatch.._batch_decorator)callablerrrrr)rrUrVrWrr)rr*r}Js=Rr"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)NrrrF)Gr'rloggingrv collectionsr dataclassesr functoolsrinspectrrtypingrrr r r r r rrrrrrrrrayrray._common.signaturerrrray._common.utilsrray.serve._private.constantsrray.serve._private.utilsrray.serve.exceptionsrray.util.annotationsr getLoggerrl StopIterationrrrr+r.rSrTrrrr@rrrrrrr}r>rr(rrr)r)r)r*s   D         W 0