`i ddlZddlZddlmZddlmZddlmZddlm Z ddl m Z m Z ddl Z ddl mZddlmZd d lmZd d lmZd d lmZd d lmZmZmZddlmZddlmZmZmZm Z m!Z!ddl"m#Z#m$Z$m%Z% d de j&de j&de j&de'ddf dZ(eGddZ)eGddZ*eGddZ+GddZ,dS)!N) dataclass)partial)count) perf_counter)OptionalUnion)nn)tqdm)PretrainedConfig)GenerationConfig)logging)ContinuousBatchProcessorMetrics attach_tracertraced)PagedAttentionCache)GenerationOutput RequestState RequestStatusget_device_and_memory_breakdownlogger)SCHEDULER_MAPPING FIFOScheduler Schedulerattention_maskcumulative_seqlens_qcumulative_seqlens_ksliding_windowreturncjtj|jj}t t |dz D]}||dz||z }||dz||z }||kr|dkr ||z dz}nd}t ||||dz} t ||||dz} tj|d| | fj||j|j } tj | |} |dkr!||z |z } | tj | | z } | |d| | f<dS)uBuilds an attention mask inplace using the cumulative seqlens of the query and key. If given a sliding window, it will also apply a sliding window mask on top. The attention mask is not boolean, it uses zeroes and -inf (or its equivalent) so it's more of an attention score bias tensor. The attention mask is a block-diagonal matrix, with each block an attention mask for a single query-key pair. Each of those block is built from a causal mask and, if there is a sliding window, a sliding window mask. An example is represented below, with seqlen_k = 8, seqlen_q = 4 and sliding_window = 6: CAUSAL MASK: █ █ █ █ █ ░ ░ ░ █ █ █ █ █ █ ░ ░ █ █ █ █ █ █ █ ░ █ █ █ █ █ █ █ █ SLIDING WINDOW MASK: ┌──────────────────────── seqlen_k - seqlen_q - sliding_window = 8 - 4 - 6 = -2 offset to the right <─┴─> ░ █ | █ █ █ █ █ █ █ █ ░ ░ | █ █ █ █ █ █ █ █ ░ ░ | ░ █ █ █ █ █ █ █ ░ ░ | ░ ░ █ █ █ █ █ █ ATTENTION MASK (sum of causal and sliding window masks): █ █ █ █ █ ░ ░ ░ █ █ █ █ █ █ ░ ░ ░ █ █ █ █ █ █ ░ ░ ░ █ █ █ █ █ █ Another example with seqlen_k = 5, seqlen_q = 3 and sliding_window = 2: CAUSAL MASK: █ █ █ ░ ░ █ █ █ █ ░ █ █ █ █ █ SLIDING WINDOW MASK: ┌──────────────────────── seqlen_k - seqlen_q - sliding_window = 5 - 3 - 2 = 0 offset to the right <┴> | ░ █ █ █ █ | ░ ░ █ █ █ | ░ ░ ░ █ █ ATTENTION MASK (sum of causal and sliding window masks): ░ █ █ ░ ░ ░ ░ █ █ ░ ░ ░ ░ █ █ r.dtypedevice)diagonalN) torchfinfor#minrangelenslicefullshaper$triutril)rrrr min_valueiseqlen_qseqlen_kcausal_diagonal query_range key_range minus_infmaskedsliding_diagonals /home/jaya/work/projects/VOICE-AGENT/VIET/agent-env/lib/python3.11/site-packages/transformers/generation/continuous_batching/continuous_api.pybuild_attention_maskr;%sxt N0115I 3+,,q0 1 1=='A.1Ea1HH'A.1Ea1HH h  8q==&1A5OOO035I!a%5PQQ .q13GA3NOO J 3 Y6 7 =  &!(    I@@@ A  '(2^C  ej5EFFF FF6<sK233-==ceZdZUejed<eejed<ejed<ejed<ejed<eed<eed<eejed<eejed <ejed <e ed <d Z e ed <dS)PagedAttentionArgs input_idsr position_idsrr max_seqlen_q max_seqlen_k write_index read_indexlogits_indicescacheF use_cacheN) __name__ __module__ __qualname__r&Tensor__annotations__rintlistrrGboolr<r:r>r>ys|U\****,,&&&,&&&el####U\""""L   Itr<r>ceZdZ d-dedededejdejdej d e j d e j d e d ed ededdfdZeddeddfdZdefdZee jd.defdZdefdZdZedZedefdZedefdZedeedeed eeed!eeed"eed#eeeee eeffd$eeddfd%Z!ed&Z"eded'efd(Z#ed)Z$edefd*Z%ed+Z&ed,Z'dS)/ContinuousBatchProcessorFTrFconfiggeneration_config input_queue output_queue stop_event model_device model_dtype scheduler streamingmanual_eviction slice_inputsr Nc ||_||_||_||_||_||_||_||_| |_| |_ | |_ | |_ t|dddn|j |_ g|_|j|_t!|j|_d|_d|_d|_||jdS)aInitialize the continuous batch processor. Args: cache: A [`PagedAttentionCache`] object config: The model configuration generation_config: The generation configuration input_queue: Queue for incoming requests output_queue: Queue for outgoing results stop_event: Event to signal processing should stop model_device: Device for model inputs/outputs model_dtype: Data type for model inputs/outputs scheduler: The [`Scheduler`] to use streaming: Whether to stream tokens as they're generated manual_eviction: Whether to manually evict blocks from the cache slice_inputs: Whether to slice the inputs to the model rNrr)rFrSrTrUrVrWrXrYrZr[r\r]getattrrrequests_in_batchmax_batch_tokensrmetricstotal_query_lengthtotal_key_lengthtotal_batch_sizesetup_static_tensors num_groups) selfrFrSrTrUrVrWrXrYrZr[r\r]s r:__init__z!ContinuousBatchProcessor.__init__s>  !2&($(&"".($+63CT#J#J#RaaX^Xm57!& 66u7MNN #$ ! ! !!%"233333r<) standalonergc jjjjjztjjd_t jdffij_ t jdffij_ t jdzffij_ d_ t jffij_ t jdffij_tjdd}| tjdd}|dvrdgndg}t#t%|}fd |D_t(|d_r+ddzfjjd fd |D_nd_fd t5|D_fd t5|D_ddS)Nr"rr layer_typesr)rNfull_attentionsliding_attentioncFi|]}|tjdzfijSrr&emptytensor_metadata).0 layer_typeTrhs r: zAContinuousBatchProcessor.setup_static_tensors..sA% % % ISJ QUDDt/CDD% % % r<)sizer#r$c4i|]}|tjdiS)rP)r&rr)rtruattn_mask_kwargss r:rwzAContinuousBatchProcessor.setup_static_tensors..s-"m"m"mS]:u{/N/N=M/N/N"m"m"mr<c@g|]}tjffijSrPrq)rt_rvrhs r: zAContinuousBatchProcessor.setup_static_tensors..s1#i#i#iRSEK$M$M8L$M$M#i#i#ir<cDg|]}tjzfijSrPrq)rtr|rv num_pagesrhs r:r}zAContinuousBatchProcessor.setup_static_tensors..s3"s"s"s\]5; A #W#W$BV#W#W"s"s"sr<T) full_reset)rarF num_blocks block_sizer&int32rXrsrrr?r@rrArE output_idsr_rSrNsetrdictfromkeysrBreturn_attention_maskrYrr)write_index_storageread_index_storagereset_static_tensors)rhrgrlrrvrzrs` @@@r:rfz-ContinuousBatchProcessor.setup_static_tensorssk  !J)DJ,AA ).@QRRaVDDt/CDD!KAGG$2FGG$)KQ$Q$QD)0K0K+,,ReQfK3{++,, % % % % % Wb% % % !!MM+q99  % % ' ' 'Aq)a-0)+    #n"m"m"mal"m"m"mD  "&D $j#i#i#i#iW\]gWhWh#i#i#i "s"s"s"s"s"safgqarar"s"s"s !!T!22222r<c"|jjdkS)Npaged_attention)rS_attn_implementationrhs r:rz.ContinuousBatchProcessor.return_attention_masks{/3DDDr<rc|s|jsa|jdd}|jdd}|jdd}n|j}|j}|j}|jddd|f|j ddd|f|j d|dzd|_ |j d| d|jddd|f d|jD]}|j|d|dzd|j|<|jN|j|ddddd|d|f t%j|jjt-|jjD]U}|j|d| d|j|d||z dVdS)zReset static tensors for the next batch. In between batches, reset only the parts that were used in the last batch, but for initialisation, we can reset everything using the (full_reset) flag.rNr)r]rrxrrcrdrer?zero_r@rrArEfill_rrrBrr&r'rYr(r)rFrg)rhrq_lenk_lenb_sizerur1s r:rz-ContinuousBatchProcessor.reset_static_tensorss[  +T. +,Q/44R88E+A.33B77E-a055a88FF+E)E*F qqq&5&y!''))) !!!VeV)$**,,, !,FQJ,/55777 FUF#))"--- 6E6 "((,,,3 o oJ  %j 1,FQJ, ? E E G G G,-D j )".#J/111fuffuf0DEKKEKX\XhLiLiLmnnntz,-- B BA  $Q ' / 5 5b 9 9 9  #A & 7 = =b A A A A B Br<c v|jr|jn|jdd}|jr|jn|jddz }|jddd|f|jddd|f|jd|dz|j|j d|iii|j |j |j dd }t|j}t!|dkr|jD]\}}|d|dz|d|<|j||d|<|jR|jr||n|j|d}|j|d d|d|f|d |<n|d}|j|d|dz|d<|j||d<|jY|j||}|jr|n|j|d}|j|d d|d|f|d <|jd|d <|S) z2Get model keyword arguments for the current batch.rrrNF) r?r@ cu_seq_lens_qrArE cu_seq_lens_krBrrDrCrFrGrrB.r)r]rcrrxrerr?r@rArErDrCrFrNrkeysr*itemsrBr)rhrrkwargsrlru seqlens_krs r:get_model_kwargsz)ContinuousBatchProcessor.get_model_kwargss,0+<f''$BZ[\B]BbBbceBfBf*.*;g&&AZA_A_`bAcAcfgAg6E6 2 -aaa%i8!6|!|D -"1&5&9 /+Z    4499;;<< {  a  )-)B)H)H)J)J p p% I6? &1* 6M' 3595Fz5R~&z2&2151BpIf--H[\fHgHlHlmoHpHpE;?;Nz;Z[^`faf`fhninhn[n;oF+,Z8  p%QJ&*&? &KLfWXjL&YF? #%)%6z%BF> "".1*=fE!%!2`8KJ8W8\8\]_8`8`+/+>z+J3PVQVPVX^Y^X^K^+_'(   &'+F# $ r<c d|jd|jd|jjd|jjd |zS)Nz%ContinuousBatchProcessor(input_queue=z, output_queue=z, active_requests=z, waiting_requests=))rUrVrZactive_requestswaiting_requestsr__repr__rs r:rz!ContinuousBatchProcessor.__repr__Es uD4D u uUYUf u u#~= u uRVR`Rq u u u##%%..00 1 r<c|js |j}|6|j|nz#t j$rYdSt$r\}tj d|dt d}|| ||Yd}~nd}~wwxYw|jdSdS)z?Pull new requests from the input queue and add to waiting list.NzError processing new request: Texc_infostate) rUrr get_nowaitrZadd_waiting_requestqueueEmpty Exceptionrerrorlocalsget_handle_request_error)rhres r:_get_new_requestsz*ContinuousBatchProcessor._get_new_requestsLs"((** 9 9(3355=2259999;    9 9 9 AaAADQQQQ&,hhll7&;&;$..q%888  9"((** 9 9 9 9 9s#AAC $ C -ACC rc~tj|_t||_t |jtr%|j|j|_ ng|_ |j |j |j|j |dS)z(Handle general request processing error.N)rFAILEDstatusstrr isinstance request_idrZ!get_active_request_static_outputsstatic_outputsrbrecord_request_completion created_timerVputto_generation_output)rhrrs r:rz.ContinuousBatchProcessor._handle_request_error^s%+ %jj  e& , , &#'>#S#STYTd#e#eE #%E  ..u/A5CSTTT e88::;;;;;r<c* ||j|jsdS|jt |jjt |jj|j |j |_ |j sdS|j |j | d|_d|_d|_g}g}dg}g}t#|jt&rd|jDndgdt)|jjD}dt)|jjD}|j D]}|j}t |j} |j|j|| } |xj| z c_|xjt7| z c_|xjdz c_|xj| z c_||j|t)||| z||d| zt7|j| |_|j s||ddz | !D]R\} } | | d| zt7|j"| | |j"| <S|j#|j|| ||j$|j|| ||%|||||||j&|jtOj(tRj*rt#|jt&r!t7fd|jD} nd} tOj+d t |j d t |jjd t |jjd |dd | d|j, dS)zPrepare tensors and metadata for the next model forward pass. Returns True if there are requests to process, False otherwise.Frci|]}|dgS)rrP)rtrus r:rwz?ContinuousBatchProcessor.prepare_next_batch..s#`#`#` J#`#`#`r<cg|]}gSrPrPrtr|s r:r}z?ContinuousBatchProcessor.prepare_next_batch..s???Qb???r<cg|]}gSrPrPrs r:r}z?ContinuousBatchProcessor.prepare_next_batch..s@@@ar@@@r<rrc34K|]}|dVdS)rNrP)rtrurs r: z>ContinuousBatchProcessor.prepare_next_batch..s.jj*-j9"=jjjjjjr<z Scheduled: z , Waiting: z , Active: z . cum Q: z . cum KV: z, free blocks: T)-rrZclear_cancelled_requestshas_pending_requestsrbrecord_queue_metricsr*rrschedule_batchrar`record_batch_metricsrrcrdrerrrr)rFrgposition_offset prompt_ids get_seqlens_krmaxvaluesextendappendrAremaining_prompt_idsrrBextend_read_indicesextend_write_indices_build_tensorsrecord_kv_cache_memory_metricsr isEnabledForrDEBUGdebugget_num_free_blocks)rhr?r@rrErDrCr past_length query_lengthrrulayer_type_seqlen_kckrs @r:prepare_next_batchz+ContinuousBatchProcessor.prepare_next_batchms     //111~2244 5 ))#dn.L*M*MsSWSaSrOsOsttt"&!>!>t?T!U!U% 5 ))$*@AAA !!####$ ! !  !s d/ 6 6 '#`#`dF_#`#`#` $%3 ??% (=">">??? @@5)>#?#?@@@ + f fE/Ku/00L 001A;P\]]I  # #| 3 # #  ! !S)9)9););%<%< < ! !  ! !Q & ! !  ! !\ 1 ! !   U- . . .   k;3M N N O O O ' '(&BCCC3>QQI]^`IaQQQQ.2j.L.L.N.NQQ    tr<r?r@rDrCrrrEcttjfi|j}|||jdddt |f<|||jdddt |f<|||jdt |<|||jdt |<| D]_\} } || |j | dt | <|j +t|j | || | dkr|j nd`g|_g|_t!t#||D]\} } } || |j| dt | <|| |j| dt | <|jrt | n|j| d}|jrt | n|j| d}|j|j| d||j|j| d|dS)zeBuilds the actual tensors for the current batch, by modifying the already allocated tensors in place.Nrnr)rrrrr)rr&tensorrsr?r*r@rrErrrr;rrDrCziprrrr]rxr)rhr?r@rDrCrrrE to_tensorrulayer_type_seqlens_kr1group_read_indicesgroup_write_indicesrws r:rz'ContinuousBatchProcessor._build_tensorssELAAD,@AA /8i .B.Bqqq*C NN**+4=Il4K4K!!!0s<00001AJK_A`A`!"=C(<$=$="=>5>Y~5N5N1c.11121E0J0J0L0L   ,J,QZQZ[oQpQpD %j 12MC8L4M4M2M N".$#'#6z#B)=)=:DH[:[:[4#6#6ab :=eggzS^:_:_ E E 6A!#6DMIN`DaDaD #A &'@-?)@)@'@ AFOiPcFdFdD $Q '(B#.A*B*B(B C+/+<e&'''$BYZ[B\BaBabdBeBeA,0,=g'(((4C[\]C^CcCcdfCgCgA O " "4#:1#=bqb#A B B B   # #D$ @   ! !%"<"<">"> ? ? ? ? ? \]3 3 3   ! !%"<"<">"> ? ? ? ? ?4 3r<c |}g}t|jD]*\}}|j}t |jdkr|j|j|jtj |_ ||j |}|g|_ ||ra|j|j|j|j|j|j ||||||j tjkrtj|_ ,|jdkrt3ddS)z0Update request states based on generated tokens.r)evict_from_cachezNo more free blocksN)r enumerater`rr*rrbrecord_ttft_metricrrDECODINGrrErupdate_with_tokenrrZfinish_requestr\rrPREFILLING_SPLITSPLIT_PENDING_REMAINDERrFr ValueError)rh out_tokensfinished_request_idsr1rreq_idrs r: update_batchz%ContinuousBatchProcessor.update_batch sjZZ\\ !!$"899 E EHAu%F5-..!33 //0BEDTUUU,5 "4#6q#9:$)7 **5118L::5;MuO_```N11%2BZ^ZnVn1ppp(//777''u5555!???,D : ) ) + +q 0 0233 3 1 0r<c4|jS)z2Check if there are any active or waiting requests.)rZrrs r:rz-ContinuousBatchProcessor.has_pending_requests!s~22444r<c|j}|D]7}||||j|j8dS)z&Handle errors during batch processing.N)r`rrZrr)rhr failed_reqsreqs r:handle_batch_errorz+ContinuousBatchProcessor.handle_batch_error&sV,  : :C  & &uc 2 2 2 N ) )#. 9 9 9 9 : :r<ct|jj}|D]7}||||j|j8t|jjD]7}|jj |}|||8|jj dS)zFail all active requests with the given error. Args: error: The error to report in the failure message N) rNrZrrrrrrrpopwaiting_requests_orderclear)rhrrequestsrrs r:fail_all_requestsz*ContinuousBatchProcessor.fail_all_requests.s6==??@@ < :??AABB 5 5FN377??E  & &ue 4 4 4 4 -3355555r<)FFT)F)(rHrIrJrr r rQueue threadingEventr&r$r#rrOrirrMrfrno_gradrr>rrrrrrrNrrrrrrrrrr rPr<r:rRrRs} %!9494"94!94, 94 [ 94 k 94O94l94[9494949494 94949494v Vt)3s)3t)3)3)3)3VEtEEEE U]__BBtBBB_ VBB*"4****X    99 V9"  <, < < < V < [D[[[ V[z *E9*E3i*EcO *E $s)_ *E #3i *E$DItCcN/C$CD*ES *E *E*E*E V*EX  V @ @S@@@ V@ 44 V4* 5d555 V5 :: V: 66 V666r<rRc eZdZdZ d/dedededefd Zed Zd Z d0d ede e fdZ d1de e fdZ d2deede ede edefdZdeeefdZdefdZd2de efdZdZdZedeefdZedefdZedZedefd Zed!"d#Zed$"d%Z ed&"defd'Z!d(Z"ed)"defd*Z#ed+"d,Z$ede efd-Z%edefd.Z&d S)3ContinuousBatchingManagerzManager for handling continuous batching of generation requests. This class provides the user interface for submitting generation requests, retrieving results, and managing the background generation thread. FrTrTr\r[r]c||_||jn|}||_tj||_tj|_tj|_ ||_ t|dd|_ d|_ d|_tj|_d|jj_t|dd|_|j||_t|dd|_t|d d|_||_d|_||_|jrt5d dS) aSInitialize the continuous batching manager. Args: model: The language model for generation generation_config: Configuration for generation parameters max_queue_size: Maximum size of the request queue (0 = unlimited) streaming: Whether to stream tokens as they are generated N)maxsizelog_prob_generationFr do_sampleTuse_cuda_graphprofilez!Cuda graphs are not supported yet)evalmodelrTrr rUrVr r rWr[r_r_generation_thread_request_counterLock _request_locktop_pr_get_logits_processorlogit_processorrrr\batch_processorr]NotImplementedError)rhrrTr\max_queue_sizer[r]s r:riz"ContinuousBatchingManager.__init__MsB"ZZ\\ 7H7PE33Vg!2 ;~>>>!KMM#/++"#*+<>SUZ#[#[ "& !&^---1 $* !2KFF#z??@QRR%&79I5QQ0)UCC .CG(   K%&IJJ J K Kr<c|j/|jrtjddSt j|_tj|j |_|j dS)z'Start the background generation thread.Nz"Manager thread is already running.)target) ris_aliverwarningrr  _result_queuer Thread_run_generation_loopstartrs r:r)zContinuousBatchingManager.startusy  " .43J3S3S3U3U . N? @ @ @ F"[]]"+"2$:S"T"T"T %%'''''r<cF|jduo|jS)z5Check if the background generation thread is running.N)rr$rs r: is_runningz$ContinuousBatchingManager.is_runnings$&d2Yt7N7W7W7Y7YYr<Nblocktimeoutc|jtjddS|js-|jtjd|r||dSdS)zSignal the background thread to stop. Args: block: Whether to wait for the thread to stop timeout: Maximum time to wait for the thread to stop NzManager not started.z'Stopping continuous batching manager...)rrr%rWis_setrinfojoin)rhr,r-s r:stopzContinuousBatchingManager.stops  " * N1 2 2 2 F%%'' C O   ! ! ! KA B B B   IIg       r<c|jg|j||jrtjddStjdd|_dSdS)zWait for the background thread to finish. Args: timeout: Maximum time to wait for the thread to stop Nr-z2Generation thread did not exit after join timeout.z$Continuous Batching Manager stopped.)rr1r$rr%r0)rhr-s r:r1zContinuousBatchingManager.joins|  " .  # ( ( ( 9 9 9&//11 /STTTTT BCCC*.''' / .r<r?rmax_new_tokensr cv|9|j5d|j}|xjdz c_dddn #1swxYwY| |jjn|}t |t |t |||jj}|j|ddtj d|d |S) a/Add a new generation request to the queue. Args: input_ids: Input token IDs to use as prompt request_id: Optional custom request ID (auto-generated if None) **kwargs: Additional generation parameters Returns: str: The request ID Nreq_r)rrfull_prompt_idsr5 eos_token_idT r,r-zAdded request z to queue.) rrrTr5rrNr9rUrrr)rhr?rr5rs r: add_requestz%ContinuousBatchingManager.add_requests  # + +;D$9;; %%*%% + + + + + + + + + + + + + + +CQBX/>>^l!I OO)/<     U$;;;  sdpa_paged eager_pagedflash_attention_2rPrPr<r:#supported_attention_implementationsz=ContinuousBatchingManager.supported_attention_implementationssAAAAr<cdS)NrPrPrPr<r: default_attention_implementationz:ContinuousBatchingManager.default_attention_implementations|r<ctj|jj}|tjtj|5||dddn #1swxYwYtj|tj |_ tj |j |5||ddddS#1swxYwYdS)N)r$)stream) r&cudaStreamrr$ wait_streamcurrent_streamrW_generation_step CUDAGraphgraph)rhrrWs r:warmupz ContinuousBatchingManager.warmups""$**;"<<5:4466777 Z  v & & 3 3  ! !/ 2 2 2 3 3 3 3 3 3 3 3 3 3 3 3 3 3 3  !!##//777Z))++ Z  dj  8 8 3 3  ! !/ 2 2 2 3 3 3 3 3 3 3 3 3 3 3 3 3 3 3 3 3 3s$;BB!$B!"EE  E rcJ|}tj5||}|jr|j||||}|||ddddS#1swxYwYdS)z6Perform a single generation step. This is cuda graphedN) rr&r _model_forwardr output_probscopy__process_logit_sample)rhr batch_datalogitsprobss r:r\z*ContinuousBatchingManager._generation_steps%5577 ]__ 1 1((44F' ;,226:::'' F;;E LL% 0 0 0  1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1sA#BBB model_forward) span_namec&|jdi|jS)NrP)rrg)rhrfs r:raz(ContinuousBatchingManager._model_forwardstz''J''..r<logit_processingcXt|jdr'|j|d|d|j\}}}|||z|}|d||z}|||}||||S)Nset_continuous_batching_contextrErr?)hasattrrrnr-view) rhrfrg batch_sizeseq_len vocab_size logits_2d input_ids_2dprocessed_logits_2ds r:rdz(ContinuousBatchingManager._process_logits 4')J K K   @ @+,j.I    +1,' GZKK W 4jAA !+.33J4HII #22<KK#'' GZHHHr<samplingc|jrftj|d}t j|ddd}|d}nt j|d}| d}|j ddd|f |dS)Nr)dimrr) num_samples) rr functionalsoftmaxr& multinomialsqueeze unsqueezeargmaxrxrrc)rhrrh next_tokenstokenss r:rez!ContinuousBatchingManager._sample2s > 6M))%R)88E+E!H!DDDLLRPPK%//22KK,u"555K!!!$$"111gvg:.44[AAAAAr<c d} t}t|jj|j|jj|jjt|jdd}tj dt|z dd}t|jdrAtj |jj d}|tjd|dt}nt}t}t!||jj|j|j|j|j|jj|jj|||j|j|j|j }||_d |_tj d t|z d|jr|rR|||xjd z c_|j>|RnF#t8$r9}tjd |d|||Yd}~nd}~wwxYwtjddS#tjdwxYw)z6Main processing loop running in the background thread.N_tp_size)tp_sizezPagedAttentionCache created in z secondsrZz Scheduler 'z ' not found. Defaulting to FIFO.)r]rzbatch_processor created in rzError in generation loop: TrzGeneration loop finished.) rrrrSrTr$r#r_rrrorrrZr%rrRrUrVrWr\r[r]r current_batchr/r_inner_generation_looprr_handle_critical_errorr0)rhrref_timepaged_attention_cacherZrs r:r(z.ContinuousBatchingManager._run_generation_loop@s/ 5#~~H$7 !& !   J== %%% ! L^<>>H;T^^^ _ _ _It-{;; *-1$2H2RTXYY $N#\#\#\#\]]] -I* #~~H6% !& ! !   /1EFF$!.   O$3D !"D  LZ|~~7PZZZ [ [ [--// (O4X4X4Z4Z (++O<<<""a'""--// (O4X4X4Z4Z ( < < < L9a99D I I I I  ' '? ; ; ; ; ; ; ; ; < K3 4 4 4 4 4FK3 4 4 4 4s0H HI- I/I I-II--Jgeneration_loopc tjrtj|sdSt jtjkr3t\}}}}t j d|d|d|d|tjr|j r|j dkr| |nt|dr\ |nq#t $r9}t jd|d ||Yd}~dSd}~wwxYw||n||tjrtj|dS) Nz[Memory] Device: z , Total: z , Reserved: z , Allocated: rr^zModel forward pass failed: Tr)r&rX is_available synchronizerrlevelrrrrrrr_ro _graph_replayrrrr\r)rhrr$totalreserved allocatedrs r:rz0ContinuousBatchingManager._inner_generation_loopts : " " $ $ % J " " $ $ $1133  F <7= ( (1P1R1R .FE8Y LsVssessQYsshqss t t t : " " $ $ 3)< 3!Q&& O,,,,w'' 7&&(((( L!Bq!B!BTRRRR#66q999FFFFF %%o6666  ! !/ 2 2 2 : " " $ $ % J " " $ $ $$$&&&&&s7D E.E  E graph_replayc8|jdSr?)r^replayrs r:rz'ContinuousBatchingManager._graph_replays r<c|j |j}||||2#t j$rYnwxYw|||dSdS)z:Handle critical errors that terminate the generation loop.TN)rWrrUrrrrr )rhrrreq_datas r:rz0ContinuousBatchingManager._handle_critical_errors   K+6688".#99%JJJ K{    D   &  - -e 4 4 4 4 4 ' &s3AA A c|jstd|j!|jj|dSdS)zSEvict a request from the cache. It is assumed that the request is already finished.z0Manual eviction is not enabled for this manager.N)r\ RuntimeErrorrrZrrCs r:evict_request_from_cachez2ContinuousBatchingManager.evict_request_from_cachesP# SQRR R   +  * 9 9* E E E E E , +r<)FrTT)FNr?)NN)'rHrIrJ__doc__r rOrirr)r+rfloatr2r1rNrMrr<r@rDrrGrJrN staticmethodrrSrUr_rRr\rardrer(rrrrrPr<r:rrEs!&!&K&K,&K &K  &K&K&K&K&KP (( V(ZZZ$%$ / /HUO / / / /gk  c 08  V^_bVc      D24S ?2222PPPPP8DT;U*dddBSBBB\Bc\  3 3 V 3 10H111 V1 Vo&&&//'&/ V()))II*)I& Vj!!! B'? B B B"! B252525h V'((('6N''')('4 Vn%%%&% 5XF^=_555 V5$ F3FFF VFFFr<rceZdZdZ ddeededed ed ed ef d Z e e j dd e e edeeded ed e e ef dZdS)ContinuousMixinz?Mixin class for models to add continuous batching capabilities.NFrTrTr\r!r[r]r c&t|dr t|drt|dstd||n|j}|td|jt jdd|_t|||||| S) aInitialize a manager for continuous batching inference. Args: generation_config: Custom generation configuration max_queue_size: Maximum size of the input request queue streaming: Whether to stream tokens as they are generated Returns: `ContinuousBatchingManager`: The manager instance to add requests and retrieve results. rSr$r#z;Model must have 'config', 'device', and 'dtype' attributes.Nz8A GenerationConfig must be provided or set in the model.zE`eos_token_id` not set in GenerationConfig. Setting to -1 (disabled).r)rrTr\r!r[r])roAttributeErrorrTrr9rr%r)rhrTr\r!r[r] gen_configs r:init_continuous_batchingz(ContinuousMixin.init_continuous_batchings$tX&& `gdH.E.E `WUY[bMcMc ` !^__ _*;*G&&TMc  WXX X  " * Nb c c c&(J #)(+)%     r<r= progress_barc |sgStjtjkrtjdd}|||}|i}t|} ddlm } | tg5t|| d|dd 5} |j |fi|d} | |kr| d } | r<| j } | jtjkr| || <| d z } | d n)|stjd n| |kd d d n #1swxYwYd d d n #1swxYwYn0#t($r#}tjd|dYd }~nd }~wwxYw|ddn#|ddwxYw|S)a=Generate sequences for a batch of prompts using continuous batching. Args: inputs: List of input token sequences (prompts) generation_config: Optional generation configuration **kwargs: Additional generation parameters Returns: `list[list[int]]`: A list containing the generated sequences (including prompt tokens if not handled otherwise) for each input prompt, in the same order. Returns an empty list `[]` for requests that failed. z=Progress bar is disabled when logger level is less than DEBUGF)rTr]r)logging_redirect_tqdmzSolving z requestsrequest)rdisabledescunitrr4z*Generation thread terminated unexpectedly.NzError during batch generation: Trg@r;)rgetEffectiveLevelrrr%rr)r*tqdm.contrib.loggingrr r@rGrrrrupdater+rrr2)rhr=rTrr]rmanagerresults num_requestsrpbarfinished_countrFrrs r:generate_batchzContinuousMixin.generate_batchs, I  # % % 6 6 NZ [ [ [ L//BSbn/oo 6{{  2 B B B B B B&&x00 & &&!--;L;;;" & (G(::6:::%&N(<77!(!3!3A!3!>!>! &%+%6F%} 0FFF28 .! 3 $ A#*#5#5#7#7& & -Y Z Z Z %)<77&&&&&&&&&&&&&&& & & & & & & & & & & & & & & &, O O O L>1>> N N N N N N N N N O LLtSL 1 1 1 1GLLtSL 1 1 1 1sy9E2E&)BE E&E E&E E& E2&E**E2-E*.E21F:2 F<FF:FF::G)NFrFT)NTT)rHrIrJrrr rOrMrrrr&inference_moderNrrPr<r:rrsII9= %! % % #$45% %  %  %  %  #% % % % N U9=!! <<T#Y<$$45< <  < d3i<<< V<<<r<rrp)-rr  dataclassesr functoolsr itertoolsrtimertypingrrr&r r configuration_utilsr generation.configuration_utilsr utils.loggingr utils.metricsrrrrFrrrrrrrrZrrrrKrMr;r>rRrrrPr<r:rs  !!!!!!"""""""" 333333>>>>>>$$$$$$SSSSSSSSSS&&&&&&llllllllllllllBBBBBBBBBB Q=Q=LQ=,Q= ,Q= Q=  Q=Q=Q=Q=h           v6v6v6v6v6v6v6v6t fFfFfFfFfFfFfFfFR hhhhhhhhhhr<