^jddlZddlZddlZddlZddlmZddlmZmZddl m Z m Z ddl m Z ddlmZddlZddlmZddlmZdd lmZd d lmZd d lmZmZd d lmZddlmZddlmZddl m!Z!ddl"m#Z#ddl$m%Z%m&Z&ddl'm(Z(m)Z)ddl*m+Z+ddl,m-Z-ddl.m/Z/m0Z0m1Z1m2Z2ddl3m4Z4m5Z5m6Z6ddl7m8Z8m9Z9 GddejtZ;GddZ<Gdd Z=Gd!d"Z>Gd#d$Z?Gd%d&Z@y)'N)abstractmethod)Callable Generator)contextmanager nullcontext) perf_counter)Any)nn)tqdm)logging_redirect_tqdm)PretrainedConfig)ContinuousBatchingConfigGenerationConfig)logging)LogitsProcessorList)PagedAttentionCache)%ContinuousBatchingLogitsProcessorList)DistributedHelper)"resolve_continuous_batching_config%update_cb_config_after_cache_creation)ContinuousBatchingAsyncIOsContinuousBatchingIOs) ModelRunner)OffloadingManager)GenerationOutput RequestState RequestStatuslogger)SCHEDULER_MAPPING FIFOScheduler Scheduler) WorkloadHints drain_queueceZdZUeed<ej ed<ejed<ede ddfdZ ede de fd Z y) ProtoPretrainedModelconfigdtypedeviceattn_implementationreturnNcyN)selfr,s /var/www/ramen.bs-engineer-server.com/venv/lib/python3.12/site-packages/transformers/generation/continuous_batching/continuous_api.pyset_attn_implementationz,ProtoPretrainedModel.set_attn_implementationI generation_configcyr/r0)r1r6s r2_get_logits_processorz*ProtoPretrainedModel._get_logits_processorMr4r5)__name__ __module__ __qualname__r__annotations__torchr*r+rstrr3rrr8r0r5r2r(r(Ds`  ;; LL 3 4   7G L_  r5r(c>eZdZdZd dZdeddfdZdeeddfdZy) OutputRouteraDedicated object for routing generation outputs to the right destination. When an async handler is registered for a request, the output is forwarded to that handler via ``call_soon_threadsafe``. Otherwise the output is placed on the shared ``output_queue``. r-Ncvtj|_i|_t j |_yr/)queueQueue output_queueresult_handlers threadingLock_lockr1s r2__init__zOutputRouter.__init__Zs&!KKMVX^^% r5outputc|j5|jj|j}ddd|\}}|j ||y|j j |y#1swY?xYw)zDRoute a single output to its registered handler or the output_queue.N)rHrEget request_idcall_soon_threadsaferDput)r1rKentrycallbackloops r2deliverzOutputRouter.deliver_sq ZZ @((,,V->->?E @  "NHd  % %h 7    ! !& )  @ @s &A11A:outputscLg}d}|j5|D]]}|jj|j}||\}}|j ||fC|j j |_ ddd|r||fd}|j|yyy#1swY&xYw)zRoute a batch of outputs, using a single ``call_soon_threadsafe`` to minimize cross-thread overhead. Outputs without a registered handler fall back to the shared ``output_queue``. Nc(|D] \}}||yr/r0)batchcbouts r2 _run_batchz.OutputRouter.deliver_batch.._run_batchzs$GBsGr5)rHrErMrNappendrDrPrO)r1rU callbacksrSrKrQrRr[s r2 deliver_batchzOutputRouter.deliver_batchis >@  ZZ 2! 2,,001B1BC$%*NHd$$h%78%%))&1  2 2 )!*   % %j 1 *9 2 2s A#BB#r-N) r9r:r;__doc__rJrrTlistr^r0r5r2r@r@Rs9& *.*4*2T*:%;22r5r@ceZdZdZdZdZdZdZddZdd Z d e d e ddfd Z dd Z de ddfdZ ede fdZede fdZy)BackgroundThreadStatuszTracks the status of the background thread locally and in its TP group. The status is an int that can only increase, representing how soon the thread should stop.rrrr r-Ncztj|_|j|_|j|_yr/)rFrG_local_status_lock DONT_STOP _local_status _tp_statusrIs r2rJzBackgroundThreadStatus.__init__s("+.."2!^^..r5c|j|_|j5|j|_dddy#1swYyxYw)zClear the local and TP statuses. This method should ONLY be called by the main thread itself BEFORE starting the background thread.N)rfrhrergrIs r2clearzBackgroundThreadStatus.clears9..  $ $ 0!%D  0 0 0s 9Astatus global_rankc8||j|jfvrtd|d||j5t ||j |j |_dddtjd|d|d|j y#1swY2xYw)zRequest the background thread to stop. This does not take effect immediately, only after the TP group has communicated.zInvalid stop status z from rank NzRank z3 requested background thread to stop with status = z. Now self._local_status = ) FLUSH_AND_STOP HARD_STOP ValueErrorremaxrgrhr!info)r1rkrls r2 request_stopz#BackgroundThreadStatus.request_stops $--t~~> >3F8;{mTU U  $ $ R!$VT-?-?!QD  R K= T6+Uq\`\n\n[r s  R Rs 'BBch|j5|j|_dddy#1swYyxYw)zzMark the background thread as stopped. This should be called by the main thread when the generation loop finishes.N)reSTOPPEDrgrIs r2mark_as_stoppedz&BackgroundThreadStatus.mark_as_stoppeds. $ $ .!%D  . . .s(1 tp_statusc||jkrtd|d|j||_|j5t|j||_dddy#1swYyxYw)z8Update the local and TP statuses with the new TP status.z1TP communicated a lower stop status: tp_status = z, self._tp_status = N)rhrprerqrg)r1rws r2update_with_tp_statusz,BackgroundThreadStatus.update_with_tp_statussn t &QYNRgUYUdUdThij j#  $ $ D!$T%7%7!CD  D D Ds A##A,c|jS)zVThe locally requested status, possibly ahead of the value agreed upon by the TP group.)rgrIs r2 local_statusz#BackgroundThreadStatus.local_statuss!!!r5c|jS)zKThe status last agreed upon by the TP group through a MAX-reduce operation.)rhrIs r2rwz BackgroundThreadStatus.tp_statussr5r_)r9r:r;r`rfrnrorurJrjintrsrvrypropertyr{rwr0r5r2rcrcs?INIG) 0  3  S  T  . DsDtD"c""3r5rcceZdZUeezed<eed<dedede de de de jd zd e jd zd ed ed ej"dej$dededd fdZdefdZd!dZd!dZdefdZdededd fdZdefdZd!dZdefdZ dZ!dedd fdZ"ejFde$jJdd fdZ&ejFde$jJdd fd Z'y )"ContinuousBatchProcessorinputs_and_outputs schedulercacher)r6continuous_batching_configlogit_processor input_queueN cancel_queue output_routerbackground_thread_status model_device model_dtypedistributed_helperr-c||_||_||_||_||_||_||_| |_| |_| |_ | |_ | |_ t|dd|_ |j|_|jj|j | t|dddn |j"|_|j$|_|||| | |jd}|jj&|_|j&rt)d i||_nt-d i||_t/|| |j0|j2|j*j4|j|_t9|j|j|j|j*|j|j|_y) a)Initialize the continuous batch processor. Args: cache: A [`PagedAttentionCache`] object config: The model configuration generation_config: The generation configuration continuous_batching_config: The continuous batching configuration logit_processor: The [`ContinuousBatchingLogitsProcessorList`] object used to process the logits. input_queue: Queue for incoming requests. Is None if this process is not a TP driver. cancel_queue: Queue for cancellation request_ids. Is None if this process is not a TP driver. output_router: An [`OutputRouter`] object that routes outputs to handlers or the output queue. background_thread_status: A [`BackgroundThreadStatus`] object to track the background thread status. model_device: Device for model inputs/outputs model_dtype: Data type for model inputs/outputs scheduler: The [`Scheduler`] to use distributed_helper: The [`DistributedHelper`] to use do_sampleTsliding_windowNr)rr)rr+rr)rrcpu_offload_space_gibsafety_thresholdcompute_streamr)r cb_configrrrreturn_logprobsr0)rr)rrrrrrrrrrgetattrrr set_tp_seedseedrmax_batch_tokensuse_async_batchingrrrrcpu_offload_space"cpu_offload_space_safety_thresholdroffloading_managerr model_runner)r1rr)r6rrrrrrrrrr io_kwargss r2rJz!ContinuousBatchProcessor.__init__sB  3.&(*(@%(&""4!!2KF9II ++,F,K,K\Z$+63CT#J#RaX^XmXm % 6 6*D"&#33  #'.."C"C  " "&@&M9&MD #&;&Hi&HD ##4"<"N"N7ZZ22AA#66 # ( 00nn**#66nn 00  r5cd|jd|jjd|jjd|jj j zS)Nz%ContinuousBatchProcessor(input_queue=z, active_requests=z, waiting_requests=))rractive_requestswaiting_requestsrget_model_kwargs__repr__rIs r2rz!ContinuousBatchProcessor.__repr__sj3D4D4D3EF#~~==>>QRVR`R`RqRqQrrs u%%668AAC D r5cd|_tjtjj rtjj yyr/)rgccollectr=cuda is_available empty_cacherIs r2__del__z ContinuousBatchProcessor.__del__%s8"& :: " " $ JJ " " $ %r5c|jj|jj|jj|jj y)z4Reset the batch processor for a new generation loop.N)rresetrrrfree_all_requestsrIs r2rzContinuousBatchProcessor.reset+sF %%'  %%' $$&r5c\|j7|j+t|jt|jf}nggf}t|dt|dz}|jj }|j j||\}}|jj||jjtjk(ry|dk(ry|j j|\}}|D]C} |jj|j|j j#|E|D]} |j j-| y#t$$r6}t'j(d|d|j+||Yd}~d}~wwxYw)zCommunicates with the TP group to get A. the new requests and cancellations from the TP driver, and B. an eventual stop signal from any process in the TP group. Returns True if the TP group is hard-stopping, False otherwiseNrrTFzError processing new request: exc_info)rrr&lenrr{rtp_all_reduce_stateryrwrcrotp_broadcast_object_from_rank_0r check_kwargslogit_processor_kwargsradd_waiting_request Exceptionr!error_handle_request_errorset_request_cancellation) r1payload payload_sizelocal_requested_statusrw new_states cancellationsstateerNs r2_update_tp_group_statez/ContinuousBatchProcessor._update_tp_group_state2s    'D,=,=,I"4#3#34k$BSBS6TUG2hG71:WQZ8 "&!>!>!K!K"&"9"9"M"Ml\r"s i %%;;IF  ( ( 2 26L6V6V V 1 $($;$;$[$[\c$d! M  5E 5$$11%2N2NO2259 5( @J NN 3 3J ? @  5 =aSADQ**1e44 5sAE,, F+5,F&&F+rrc8tj|_t||_t |j tr+|jj|j |_ ng|_ |jj|jy)z(Handle general request processing error.N) r FAILEDrkr>r isinstancerNr!get_active_request_static_outputsgenerated_tokensrrTto_generation_output)r1rrs r2rz.ContinuousBatchProcessor._handle_request_errorYso$++ %j  e&& ,%)^^%U%UV[VfVf%gE "%'E " ""5#=#=#?@r5c|j}|ry|jj}|D]}|jj ||jj sy|jj |j|jj\}}}}|j|jjdk(r td|jj |j|jj\}}}}|j|sy|jjr|jj|jj|tjt j"rtj$dt'|dt'|jj(dt'|jj*d|d|d |jj- |j.j1|||\}}|j2j5||j6||||j.j8 y ) zPrepare tensors and metadata for the next model forward pass. Returns True if there are requests to process, False otherwise.Frz>No requests can be scheduled and no requests can be offloaded.z Scheduled: z , Waiting: z , Active: z . cum Q: z . cum KV: z, free blocks: )requests_in_batchlogits_processorsuse_decode_fast_path num_q_tokens max_kv_read use_paddingT)rrclear_cancelled_requestsrfree_request_cpu_cachehas_pending_requestsschedule_batchrr num_pagesoffload_requests RuntimeErrorstarved_requestsrestore_scheduled_requestsr! isEnabledForrDEBUGdebugrrrget_num_free_blocksrmaybe_pad_inputsrprepare_batch_tensorsr pad_inputs)r1 hard_stoppingcancelled_statesrrrrrs r2prepare_next_batchz+ContinuousBatchProcessor.prepare_next_batchfs3 335 >>BBD% BE  # # : :5 A B~~224NR^^MjMj  ! !4::#7#7N J/{ '&&779Q>"#cddQUQ_Q_QnQn%%tzz';';R N 3\;  '! >> * *  # # 4 4 6 ::;LM   w}} - LLc"345[T^^EdEdAe@fgt~~==>?yW&-tzz7U7U7W6XZ %)$5$5$F$F|U`bv$w! k 55/"22!5%#))44 6 r5cX|jj\}}}d}g}|D]}|j}|jtj tj fvrN|jr|jr|dz }Ztd|jjd|jd|jr|jdk(rtj|_||}|||nd} |dz }|j|| } |jj!||j"| r6|j$j'|jd|j$_|j*s|jtj k(sl|j-|j/|jtj0k(s|jj!||j"|r|j2j5|gg} } |j$j6r}|j$j6j9} | j:}d| _t=|Dcgc]}| jd|}}t?||jjA| j}||z }t=|D]D}|j9}| jC}||_ |j$jE|F|dk(r|D]*}| jG||j$jH|<,|jjK| j|\}}| jM|| jM||j$j6r}| ri|jjN}|tPjRjU|n tW}|5|jjY| | dddyycc}w#1swYyxYw) z0Update request states based on generated tokens.rrzTried to update z request z in sync mode.NFz__child#)-rprepare_batch_updaterrkr FINISHEDPENDINGr has_new_tokenrnamerN generated_lenDECODINGupdate_and_check_completionr!mark_shareable_blocks_as_completecomplete_blocksrfinish_requestblock_new_requests streamingr\r PREFILLINGrr^_requests_to_forkpop num_childrenrangemincompute_max_num_forks!create_equivalent_initial_requestrforkr fork_requestextendrr=rstreamr copy_cache)r1r new_tokenslogprobscurrent_logits_indexpending_outputs future_statertokenlogprob is_finished copy_sourcecopy_destination state_to_forkrinew_request_ids num_to_forknum_to_schedule_new_request_id child_statecopy_srccopy_dstr maybe_streams r2 update_batchz%ContinuousBatchProcessor.update_batchs262I2I2^2^2`/:x - bL &&E|| 6 6 8M8MNN**#11,1,"%5ell6G6G5H RWRbRbQccq#rss))&&(A-#0#9#9EL"#78 d"&!8!89Q9QSb!c Hh   x (  # #H -/nn..4 "44CCN@N@Z5::,,^<`k`mL E %%k3CD E E +f4 E Es6P4P  P)c6|jjS)z2Check if there are any active or waiting requests.)rrrIs r2rz-ContinuousBatchProcessor.has_pending_requestss~~2244r5c|jjd}|D]M}|j||j|jj |jj Oy)z&Handle errors during batch processing.rN)rrrrrrrN)r1rfailed_future_statesrs r2handle_batch_errorz+ContinuousBatchProcessor.handle_batch_errorsa#66KKMaP0 IL  & &ul.@.@ A NN ) ),*<*<*G*G H Ir5c(t|jjj}|D]9}|j |||jj |j ;|jjt|jjjD]9}|jjj|}|j ||;|jjjy)z.Fail all active requests with the given error.N)rarrvaluesrrrNrfree_all_waiting_cpu_cachesrkeysrwaiting_requests_orderrj)r1rrequestsrreq_ids r2fail_all_requestsz*ContinuousBatchProcessor.fail_all_requestss66==?@ >::??AB 5FNN3377?E  & &ue 4 5 --335r5modelc|jj|jj}|jj |||jj y)z!Perform a single generation step.)rN)rrrr compute_batchretrieve_device_outputs)r1r! batch_datas r2_generation_stepz)ContinuousBatchProcessor._generation_stepsU ,,==$J[J[JfJf=g  ''z: 779r5c:|jj|y)aGPre-capture CUDA graphs (or trigger compile warmup) for varlen and decode paths. In async mode, both IO pairs are warmed up since each has its own graph buffer and static tensors. The varlen path is warmed up at the largest possible `(q, kv)` sizes so subsequent captures fit inside it without growing the pool.N)rwarmupr1r!s r2r(zContinuousBatchProcessor.warmups   'r5r_)(r9r:r;rrr<r$rrrrrrBrCr@rcr=r+r*rrJr>rrrboolrrrrrrrrr no_gradr Moduler&r(r0r5r2rrs-0JJJ\ "\ !\ , \ %= \ ? \ [[4'\ kkD(\ $\ #9\ ll\ [[\ \ .\  \ | # % '%%N A9 A\ Ad A>D>@KEZ5d5I6y6T6"U]]_ :bii :D : :U]]_(BII($((r5rc0eZdZdZ d,dededededzddf dZdeddfd Z d-d Z de fd Z d-d Z d.d e dedzde de ddf dZd,dededzddfdZd-dZ d/deededzdedzde de deeezdzdededzfdZ d0deeededzde de dedeef dZdeddfdZd1dedzdedzdedzfd Zd!Zdedeefd"Zded#eddfd$Ze jBd-d%Z"d-d&Z#de$fd'Z%d(e&d)e$dzddfd*Z'd(e&d)e$dzddfd+Z(y)2ContinuousBatchingManageraManager for handling continuous batching of generation requests. It provides a user interface for submitting generation requests, retrieving results, and managing the background generation thread. This class should not be created directly, but through one of the following entry points (all methods of the `ContinuousMixin` mixin): - `init_continuous_batching` - `continuous_batching_context_manager` - `generate_batch` Nr!r6rworkload_hintsr-ctj|j|_tj|_d|_t j|_t j|_ t|_ t|_d|_d|_d|_d|_d|_|j)||j+|_||_t1|dd}||nd|_t5t1|j,dd|j6|_|j8j:|_|j<r|j8j?tA|j,jC||jD|jF |_$tK|j,jL|||jHjN |_(|jPjR|_*y) ayInitialize the continuous batching manager. Args: model: The language model for generation generation_config: Configuration for generation parameters continuous_batching_config: Configuration for continuous batching parameters workload_hints: Workload hints for the continuous batching initialization (optional) )maxsizerNFnum_return_sequencesr _device_mesh) device_meshcpu_group_timeout)logits_processorper_request_processorsdrop_unsupported_processors)r)rr/has_logit_processors)+rBrCmax_queue_sizerr_request_counterrFrG _request_lockEvent_has_new_requestsrcrr@rbatch_processor_generation_thread fatal_error warmed_up_original_attn_implswitch_to_paged_attnevalr!r6rr2rr5r is_tp_driverdisable_nccl_graph_mixingmaybe_warn_nccl_graph_mixingrr8r7r8rrr) do_processingrallow_block_sharing_use_prefix_sharing)r1r!r6rr/r2s r2rJz"ContinuousBatchingManager.__init__0s !;;/I/X/XY.3kkm !&^^-!*!2)?(@%)^@D"&.2$(  !!%(ZZ\ "3&'8:PRVWOP#=#T#T(B(^(^ +M::$$0)!%!5!5!C!C + '$(#B#B#V#V r5cd|jjvrD|jj|_|jd|jjyy)zhSwitch to the paged version of the attention implementation. If the attn is already paged, does nothing.zpaged|N)r)_attn_implementationrCr3r)s r2rDz.ContinuousBatchingManager.switch_to_paged_attnusK 5<<<< <',||'H'HD $  ) )F5<<3T3T2U*V W =r5c|j|j|_|jj|jd|_y)zPre-capture CUDA graphs for varlen and decode paths by running dummy batches. Initializes the batch processor if not already done.NT)r?_create_batch_processorr(r!rBrIs r2r(z ContinuousBatchingManager.warmup{s@    '#'#?#?#AD  ##DJJ/r5cV|jduxr|jjS)zUReturns True if the background generation thread has been started and is still alive.N)r@is_aliverIs r2 is_runningz$ContinuousBatchingManager.is_runnings'&&d2Yt7N7N7W7W7YYr5c|jrtjdy|jj d|_t j|j|_ |jjy)z'Start the background generation thread.z"Manager thread is already running.N)target) rRr!warningrrjrArFThread_run_generation_loopr@startrIs r2rXzContinuousBatchingManager.startsa ??  NN? @ %%++-"+"2"2$:S:S"T %%'r5blocktimeoutkeep_for_next_session hard_stopc|jtjd|jd}|r|dz }tj|yt }|rt j nt j}|jj||jj|r|j|||sd|_n&tjd||j_|j ,|jj#|j d|_t%j&t(j*j-rt(j*j/yy)aStop the background generation thread. If the `block` flag is set to True, then this method waits for the thread to stop for a maximum time of `timeout` seconds (None means no timeout). If the `keep_for_next_session` flag is set to True, then the manager is cached on the model for future use. If the `hard_stop` flag is set, the background generation thread will be stopped immediately and pending requests will be failed.Nz% Batch processor was not initialized.zManager not started.z? Hence the unstarted manager will not be kept for next session.z:Continuous batching manager will be kept for next session.)r?r!rUr@rrcrornrrsrrljoinrrr!#_cached_continuous_batching_managerrCr3rrr=rrr)r1rYrZr[r\msgstop_trigger_time stop_statuss r2stopzContinuousBatchingManager.stops(    ' NNC D  " " *(C$XX NN3 )N:C,66I_InIn  %%22;@W@W@c@cd  II' 1%#'D  KKT U=ADJJ :  # # / JJ . .t/G/G H'+D $ :: " " $ JJ " " $ %r5rac|jy|jj||jjrtjd|dyt }tj d||z ddd|_y)zeWait for the background thread to finish. Wait can be capped using the timeout argument (in seconds).NrZz3Generation thread did not exit after join timeout (z).z+Background generation thread stopped after .2fs.)r@r^rQr!rUrrr)r1rarZends r2r^zContinuousBatchingManager.joins  " " * $$W$5  " " + + - NNPQXPYY[\ ].C KKEcL]F]^aEbbde f&*D #r5c~|jr|jdd|jjy)zTerminate the manager and release distributed resources. Safe to call multiple times. After calling this, the manager cannot be restarted.TF)rYr[N)rRrcrdestroy_cpu_comm_grouprIs r2destroyz!ContinuousBatchingManager.destroys0 ??  IIDI > 668r5 input_idsrNmax_new_tokensrrecord_timestamps eos_token_idrc |jsy|jjtjk\r0|dddddz|ddddz}t j d|dy|9|j5d |j}|xjdz c_ddd||jjn|}||jjn|}t|t||jdz ||||| } |jj!| d d |j"j%|S#1swYxYw)aAdd a new generation request to the queue. If the process is not a TP driver, this is a no-op. Args: input_ids: Input token IDs to use as prompt request_id: Optional custom request ID (auto-generated if None) max_new_tokens: Maximum number of new tokens to generate streaming: Whether to stream tokens as they're generated record_timestamps: Whether to record timestamps for each generated token eos_token_id: End-of-sequence token ID(s) logit_processor_kwargs: Keyword arguments for the logits processor. Returns: str | None: The request ID if the process is a TP driver, None otherwise. Nr z, ..., rz0Background thread is stopping. Request with ids z will be dropped.req_)rNinitial_tokensrrnrmrorrT rYrZ)rFrr{rcrnr!rUr<r;r6rmrorrar2rrPr>set) r1rlrNrmrrnrorpreviewrs r2 add_requestz%ContinuousBatchingManager.add_requests[2    ( ( 5 59O9^9^ ^"2A"- 9y~>NPQPR>^l>J>Rt--::Xd ! ?22Q6/)%#9   U$; ""$+ + +s 3%D77Einputsc t|}|j5t|j|j|zDcgc]}d| }}|xj|z c_dddt t |} |j rt| dd} |jj} | |jjjn| } | dn| } | D]\} } |jd| | |||| d| |Scc}w#1swYxYw) zgUtility function to batch `add_request` and return their IDs. Check its documentation for more details.rsNc |dS)Nrr0)xs r2z8ContinuousBatchingManager.add_requests..'s !A$r5T)keyreverserq)rlrNrmrrnror0) rr<rr;raziprKsortedr6ror!r)ry) r1rzrmrrnr num_requestsr  request_idsids_and_inputsrorNrls r2 add_requestsz&ContinuousBatchingManager.add_requestss&6{    2/4T5J5JDLaLadpLp/qr!T!:rKr  ! !\ 1 ! 2c+v67  # ##NPTUN--:: 9E9Mtzz((55S_ )1r| %3  !J D   #%-#"3)  )  +s 2 2s&D C> D>DD c|jr6|jj||jj yy)zCancel a request by its ID. If this called from a process that is not a TP driver, it's a no-op: only TP driver processes interact with the manager.N)rFrrPr>rw)r1rNs r2cancel_requestz(ContinuousBatchingManager.cancel_request9s8       ! !* -  " " & & ( r5cX|j%|jjjry |jjj d|}|5|j |k7r&|jjj |y|S#tj$rYywxYw)zRetrieve one result from the output queue. If an ID is provided, returns the first matching request. If a timeout is provided, returns None after the timeout (in seconds).NTrv) r@rrDemptyrMrNrPrBEmpty)r1rNrZresults r2 get_resultz$ContinuousBatchingManager.get_resultBs  " " *t/A/A/N/N/T/T/V ''4488tW8UF%&*;*;z*I""//33F;M{{  sABBB)(B)c#K|j\|jjrA|jd}|||j|jjr?yyyyw)z.Iterate over results as they become available.N皙?re)r@rQr)r1rs r2__iter__z"ContinuousBatchingManager.__iter__Psf%%1d6M6M6V6V6X__S_1F! %%1d6M6M6V6V6X16X1s A%A,(A,c#K|jn|jjrS|j|d}|||jry|j|jjrQyyyyw)zIterate over results matching a specific request id (blocking). Uses the shared output queue with requeue. For high-concurrency serving, use :meth:`register_result_handler` instead. Nr)rNrZ)r@rQrr)r1rNrs r2request_id_iterz)ContinuousBatchingManager.request_id_iterWsy %%1d6M6M6V6V6X__ C_HF! %%' %%1d6M6M6V6V6X16X1s A7A>:A>rRctj}fd}jj5||fjj<dddy#1swYyxYw)a+Register a callback for result delivery (streaming or non-streaming). The callback is invoked on the event loop via ``call_soon_threadsafe`` each time a result is produced for this request. For streaming requests, this happens on every token; for non-streaming, only on completion. The handler is automatically cleaned up when the request finishes. Args: request_id (`str`): The request ID to receive outputs for. callback (`callable`): Called with a ``GenerationOutput`` for each result. c||jrFjj5jjj ddddyy#1swYyxYwr/)rrrHrEr)rrRrNr1s r2 _auto_cleanupzHContinuousBatchingManager.register_result_handler.._auto_cleanupqsf V !!#''--M&&66:::tLMM$MMs 'A!!A*N)asyncioget_running_looprrHrE)r1rNrRrSrs``` r2register_result_handlerz1ContinuousBatchingManager.register_result_handlerds^'') M    % % S>KT=RD   . .z : S S Ss AA#cd} |j}||_d|_|jr@|j s t d|j |xjdz c_ |j }|r6|j |j|xjdz c_n|jjtjk(rno|jjtjk(r|jsn7|jjd|jj!t#|j$t&r8d|j$j(z |j$_|jt d|jj}|j+|||jj5t/j6d y#t,$r6}t/j0d |d |j3||Yd}~jd}~wwxYw#|jj5t/j6d wxYw) z6Main processing loop running in the background thread.Nrz$Failed to bootstrap the first batch.rTrrezdGeneration loop finished before this request completed w/ self.background_thread_status.tp_status = zError in generation loop: rzCGeneration loop finished and background thread exited successfully.)rOr? current_batchrrrr&rrrwrcrornrr>waitrjrrr current_pair_fail_all_remaining_requestsrr!r_handle_critical_errorrvrr)r1r?requests_availablerrs r2rWz.ContinuousBatchingManager._run_generation_loop|s8 _"::XYBCoFhFhFuFuBu22?,,.!wTMjMjMtMtLxyE  - -e_ E  ) ) 9 9 ; KK] ^  < LL5aS9D I  ' '? ; ; <  ) ) 9 9 ; KK] ^s*GH I ,H>9I>II1I7c||j td|jj|jy)z=Perform a single generation step. This is mostly cuda graphedNzNTried to perform a generation step before the batch processor was initialized.)r?rr&r!rIs r2r&z*ContinuousBatchingManager._generation_steps2    'op p --djj9r5ct|dd}t|tr|j|St |j j |j|j j|jt|j di|j j}|j|_ t|j|j|j|j|j dkDrd|j_|jj$}t'j(|d}|t+j,d|dt.}|||jj0|jj2 }t||j j |j4|j|j6|j8r |j:nd|j8r |j<nd|j>|j@|j j|j j||j }|S) ziCreate a new batch processor. If an already initialized batch processor exists, it is reset and returned.r?Ntp_plan)r)rr+rrr*)r num_blocksruse_prefix_sharingrz Scheduler 'z ' not found. Defaulting to FIFO.)r safety_marginmax_requests_per_batch) rr)r6rrrrrrrrrr)!rrrrrr!r)rr+rr*rrKrrrnum_sliding_attention_groupsmax_blocks_per_requestscheduler_typer"rMr!rUr#rrr6rrFrrrr)r1r?paged_attention_cacher scheduler_clsrs r2rOz1ContinuousBatchingManager._create_batch_processors"$(94@ o'? @  ! ! #" "!4::$$'+'F'F::$$#66DJJ 26**"" ! $9#K#K -55,772CC#77   ! = = AEFD + + B88GG)--ndC  NN[(88XY Z)M!'99GG#'#B#B#Y#Y 3'::$$"44'+'F'F 00,0,=,=((4.2.?.?**T,,%)%B%B** ((#66 r5rr?c||_|jjtj|j j |j jdtj|j||y)z:Handle critical errors that terminate the generation loop.)rkrlrN) rArrsrcrorrlrr)r1rr?s r2rz0ContinuousBatchingManager._handle_critical_errorsk! %%22)33AXAXAdAd 3 33A7M7W7WX ))%Ar5c |jj}||j||/#tj$rYnwxYw||j |yy)zCFail all remaining requests in the input queue and active requests.N)r get_nowaitrrBrr )r1rr?req_datas r2rz6ContinuousBatchingManager._fail_all_remaining_requestssl ++668".#99%J{{     &  - -e 4 's02AAr/r_)TNFF)NNFFN)NFF)NN))r9r:r;r`r(rrr%rJrDr(r*rRrXfloatrcr^rkrar}r>r ryrrrrrrrrrr=r+rWr&rrOrrrr0r5r2r.r.'s04 CW#CW,CW%= CW &, CW  CWJX*>X4X ZDZ( $&+ 0%0%0% $ 0%  0%  0%d +e +edl +d +9"&%)"'/3797$J7d 7  7  7DIo,7#&7 t7x&*"' !T#Y!d ! !  ! #& ! c!F))) S4Z  YilpYp  # )D  D rzrn progress_barc |siStjtjkrtjdd}| |j n|} | j | j nd} t|| z} |jdd} | | jn| } ttd|D| | nd| } |rtjd |j||d d ||| }ttg}t| | d | dd}i}d}|5}|5|5} |j|| |}|| kr|j!d}|r8|j"}|j%rM|||<|dz }|j'dn1|j)s!tj*dt-dn|| krdddddddddi}g}D],}|j1|}||||<|j3|.|rtj*d|d|S#t.$r$}tj*d|d Yd}~d}~wwxYw#1swYxYw#1swYxYw#1swYxYw)a-Generate sequences for a batch of prompts using continuous batching. Args: inputs: List of input token sequences (prompts) generation_config: Optional generation configuration continuous_batching_config: Optional continuous batching configuration record_timestamps: If set to true, the requests will have a timestamp for each token generated progress_bar: If set to true, a progress bar will be displayed persistent_manager: whether to persist the manager after the generation is finished. Default is False. warmup: whether to pre-capture CUDA graphs before processing requests. Default is True. Returns: `dict[str, GenerationOutput]`: a dictionary of request ids to GenerationOutput objects z=Progress bar is disabled when logger level is less than DEBUGFNrrmc32K|]}t|ywr/)r).0rls r2 z1ContinuousMixin.generate_batch..s!IY#i.!Isr)max_prompt_lengthmax_generated_lengthra,Since you passed `persistent_manager=True`, the manager will be kept alive after the generation is finished. However, it was sized specifically for the requests passed in `generate_batch`. If you plan to reuse the manager for a very different workload, you might want to create a new manager instead.T)r6rrYrZrr(r/zSolving z requestsrequest)totaldisabledescunit)rzrmrnrez*Generation thread terminated unexpectedly.zCReturning results of generate_batch despite unexpected termination.zError during batch generation: rz Requests z not found in results.)r!getEffectiveLevelrrrUr6r2rrrmr%rqrr r rrrNrupdaterRrprintrrMr\)r1rzr6rrnrrr(kwargsgen_cfgr2rrmr/ manager_cm logging_cmpbar_cmresultsfinished_countrpbarrrrrreordered_results missing_keyss r2generate_batchzContinuousMixin.generate_batchs4I  # # % 6 NNZ [ L->,E$((K\?F?[?[?gw;;mn6{%99  $4d;3A3I//~'!!I&!II3A3MST%   NNv ==/'A1)> +F84 %%L>3    S7J S S4 S%22!.Te3 %|3$///:F!'!2!2!--/.4GFO*a/N KKN$//1 %QRcd%|3  S S S. ! ,F[[(F!,2!&)##F+  ,  LL9\N2HI J   S >qcBTRR S' S S S S S SsaI,I IBH$:I I,$ I -I I I II I  I) %I,,I5)NNNr_)NTNNFTN)NNFTFT)r9r:r;r`rr<r=r+rr%r.rrrr*rrrrar}dictr>rrr0r5r2rrs('U]]_6:FJ/3 4 +d24 %=t$C4 &, 4 # 4 4 lAU]]_6: $FJ#(/3#"+d2#"#" #" %=t$C #" ! #"#"&,#" , -#"#"LU]]_6:FJ"'!#(m!T#Ym!,d2m!%=t$C m!  m!  m!!m!m! c## $m!m!r5r)ArrrBrFabcrcollections.abcrr contextlibrrtimertypingr r=r r tqdm.contrib.loggingr configuration_utilsrgeneration.configuration_utilsrr utils.loggingrlogits_processrrrcb_logits_processorsr distributedrinitializationrr input_outputsrrrrrrrrrr r!rr"r#r$utilsr%r&r,r(r@rcrr.rr0r5r2rs /2 63X$0&G*eL%1KKBB-,  299  ,2,2^77vg(g(V s5s5l`!`!r5