^jddlmZddlmZddlmZddlmZddlZddl m Z ddl m Z dd l mZd d lmZd d lmZd d lmZmZmZd dl mZmZmZmZmZGddeZGddZGddZGddZy)) nullcontext)partial)repeat) TypedDictN)PretrainedConfig)ContinuousBatchingConfig)get_available_devices)PagedAttentionCache)%ContinuousBatchingLogitsProcessorList) TMP_TOKEN_IDFutureRequestStatelogger)CudaGraphBufferaligned_divideattn_mask_is_neededbuild_attention_mask pad_to_pow2ceZdZUdZej ed<ej eeej fzdzed<ej ed<ej ed<ej eeej fzed<e ed<e eee fzed <e ej ed <e ej ed <ej ed <e ed <ej dzed<ej ed<e ed<y)PagedAttentionArgsaThe keyword arguments for a forward pass using paged attention, passed directly as model forward kwargs. Attributes: input_ids: Input token IDs tensor of shape `(1, total_query_tokens)`. attention_mask: Attention mask tensor or dictionary mapping layer types to masks. Can be `None` if the attention implementation doesn't require explicit masks. position_ids: Position IDs tensor of shape `(1, total_query_tokens)`. cu_seq_lens_q: Cumulative sequence lengths for queries, used for variable-length batching. cu_seq_lens_k: Cumulative sequence lengths for keys/values. Can be a tensor or dictionary mapping layer types (e.g., "full_attention", "sliding_attention") to tensors for hybrid models. max_seqlen_q: Maximum query sequence length in the batch. max_seqlen_k: Maximum key/value sequence length. Can be an int or dictionary for hybrid models. write_index: List of tensors indicating where to write new KV states in the cache, one per attention group. read_index: List of tensors indicating which cache positions to read from, one per attention group. logits_indices: Tensor indicating which positions in the output should be used for next-token prediction. cache: The [`PagedAttentionCache`] instance managing the KV cache. block_table: Block table for paged KV cache. If provided, uses `flash_attn_with_kvcache` for fused attention + cache update. More information in src/transformers/integrations/flash_paged.py logits_processor_args: List of tensors containing the arguments for the logits processors, one per request. use_cache: Whether to use caching (always `False` in continuous batching as the cache is managed externally). input_idsNattention_mask position_ids cu_seq_lens_q cu_seq_lens_k max_seqlen_q max_seqlen_k write_index read_indexlogits_indicescache block_tablelogits_processor_args use_cache) __name__ __module__ __qualname____doc__torchTensor__annotations__dictstrintlistr bool/var/www/ramen.bs-engineer-server.com/venv/lib/python3.12/site-packages/transformers/generation/continuous_batching/input_outputs.pyrrs,||LL4U\\(9#::TAA,,<<<<$sELL'8"999S#X&&ell##U\\""LL $$ <<'Or3rceZdZUdZdZeed<dedede de jde jd e d d fd Zd e d d fd Z d,ddde j j"ded d fdZe j(d,ded d fdZd-dZd ee j0eee j0fffdZde j0de j0de j0d d fdZd-dZd eeeeeee d zffdZ!deede deded ed!ed d fd"Z"d!ed efd#Z#d,d!ed e$fd$Z%d ee j0e j0e j0ffd%Z&d eed&ffd'Z'd.d(ed e j jPd zfd)Z)d*e j jPd d fd+Z*y )/ContinuousBatchingIOszA class to hold inputs and outputs for a continuous batching forward pass, using static tensors as storage. The class is meant to be self-contained, so once a set of inputs have been created, the class can be used to update the batch alone.  static_inputsr"configcontinuous_batching_configdevice model_dtypelogit_processorreturnNc&||_||_||_||_|j|_|j d|_t|dddn |j|_|j|_ d|_ d|_ d|_ t|jDcgc]}dc}|_t|jDcgc]}dc}|_d|_g|_i|_t)|_|j,|_|j0|_|j5||j7d|j8d k(r0t:j<j?|j |_ yd|_ ycc}wcc}w) a3Initialize the continuous batching I/O manager. Args: - cache: The [`PagedAttentionCache`] instance managing the KV cache. Meant to be unique. - config: The model's pretrained configuration. - continuous_batching_config: The continuous batching configuration. - device: The device to allocate tensors on. If the device is CPU, then the memory is pinned. - model_dtype: The data type for model computations. - logit_processor: The [`ContinuousBatchingLogitsProcessorList`] object used to process the logits. rsliding_windowNr F)r=T full_resetcudar;)!r"r;r9r<max_requests_per_batchcuda_graph_booleansuse_cuda_graph_varlengetattrr@return_logprobs num_q_tokens max_kv_readnum_request_in_batchrange num_groupstrue_read_sizestrue_write_sizesuse_block_tablerequests_in_batchreq_id_to_new_token_positionrgraphsread_trash_index_read_trash_indexwrite_trash_index_write_trash_index_setup_static_tensors_reset_static_tensorstyper*rCStreamcompute_streamselfr"r9r:r;r<r=_s r4__init__zContinuousBatchingIOs.__init__Ns`$   &&@&W&W#%?%S%STU%V"#*63CT#J#RaX^XmXm9II$%!+01A1A+BCaC,1%2B2B,C Dq D$;=<>)'6'8 !&!7!7"'"9"9 ""?"C ""d"3GM{{V\G\ejj//t{{/Cbf D Ds  F  Fc |jj}|jj}|j}|jj|jj z}|j jdk(xrttdkD}|j|jz}t|dzdd}tj||ftj|j ||_tj|jdftj|j |_|j%|j"|j dd|f|_|j dd|f|_|j dd|dzf|_|j d d|f|_|j d d|dzf} |j d d|dzf} |j d d|f|_i|_|jj2r| |j0d <|jj4r| |j0d<|j6rdnd} tj| |dzftj|j ||_|j8j;d|_t>jA|j0jCd|_"d|_#t>jA|j0jCd|_$tK|jLrhi|_'|j0jCD]C} tjdd|||zf|jP|j ||jN| <End|_'|jjRdkDr|nd} tj| ||jjRftj|j ||_*tj||ftjV|j ||_,tj|||zftjV|j ||_-y)aAllocates static tensors for generation inputs and outputs. This is called only once at init time, to avoid repeated allocations and enable CUDA graphs. All tensors are allocated with maximum possible sizes. The allocated tensors are: - `_bulk_input_tensor`: Storage for all the small inputs: `input_ids`, `position_ids`, `cumulative_seqlens_q`, `logits_indices`, `cumulative_seqlens_k`, `carry_over_ids`. - `attention_mask`: Optional attention masks (only for eager/SDPA implementations) - `write_index` and `read_index` storage: Cache indexing tensors for each attention group - `output_ids`: Storage for generated token IDs and maybe log probabilities if return_logprobs is True cpur )dtyper; pin_memoryrer;rNr full_attentionsliding_attention)sizerer;rf).r"rNmax_batch_tokensrE num_blocks block_sizer;r[lenr r8tensors_requiredrr*emptyint32_bulk_input_tensorlogits_processors_defaults fill_defaultsrrcumulative_seqlens_qr!carry_over_idscumulative_seqlens_knum_full_attention_groupsnum_sliding_attention_groupsrI output_idszero_total_seqlen_qr-fromkeyskeystotal_seqlen_krrrr9rr<max_blocks_per_requestr#int64write_index_storageread_index_storage)r_r=rNrorE num_pagesrf bulk_lines bulk_columns#full_attention_cumulative_seqlens_k&sliding_attention_cumulative_seqlens_knum_output_rows layer_typens r4rYz+ContinuousBatchingIOs._setup_static_tensorszs ZZ** ::66!%!@! :: / /:]D % %&6 7 :: 2 2=cD % %&9 : $33!++ .2 35;;t{{gq  .2mmD  27++Q 0)>N2NO**;;) 3##J/ #'D **;;a?JQ ;; & (I(I J++;;!  $);; ) *%++dkk^h$  #(++ %55 6ekkRVR]R]jt# r3otherstream non_blockingcD|j|_|j|_|j|_|jdd|_|jdd|_|j |_|j |_t|j|_|j|_ t|j|_ |tjj|n t}|5|jj!|j||j r(|j"j!|j"|n]|j$j!|j$||jdkDr'|j&j!|j&||j(X|j(L|j(j+D]/}|j(|j!|j(||1dddy#1swYyxYw)Nrr)rJrKrLrOrPrQrr-rrrr*rCrrrvcopy_r#rrrr)r_rrr maybe_streamrs r4_transfer_inputsz&ContinuousBatchingIOs._transfer_inputss".. ,,%)%>%>" $ 4 4Q 7!%!6!6q!9 $ 4 4#22#D$7$78!..!$"3"344:4Fuzz((0KM  w  $ $ * *4+B+BQ] * ^##!!''(8(8|'T))//0H0HWc/d##a',,2243J3JYe2f"".53G3G3S"&"5"5":":"<wJ((4::4;N;Nz;Ziu:vw w w ws 0DHHrBc|r|jjdn |j}|r|jjdn |j}|r|j dznt |j|j }|jd|jd|dzfj|r#|j|j|jdd|_ |jd|j|jddd|fj|jD]}d|j|<d|j |<|j".|j"|ddddd|d||zfj%t'j(|j*j |r|j,ddd|fj%d|jddd|fj%|j.|jddd||zfj%|j0y|j2r%|j,ddd|fj%dy|jddd|fj%|j.|jddd||zfj%|j0y)a Reset static tensors for the next batch. For efficiency, this only resets the portions of tensors that were actually used in the previous batch, using the attributes num_q_tokens and max_kv_read. If a (full_reset) is requested, the entire tensor storage is reset. r Nr)rrnrJrrKrEminrvr8rrwrr!r~r{rrrfill_r*finfor<r#rXrVrQ)r_rBq_lenkv_lenb_sizers r4rZz+ContinuousBatchingIOs._reset_static_tensorss6@((--b1TEVEV5?((--b1TEUEU4>,,q0CHYHY[_[v[vDw  4$"4"4 4k k ABHHJ <@<[<[D # #D$6$6$8 9 GV$**, 7F7 #))+33 J,-D  j )./D   +"".##J/1fuf>N>N0NOUUKK 0 0155      QZ ( . .r 2  $ $QY / 5 5d6M6M N  # #A'7'7$7 8 > >t?U?U V  ! !   QZ ( . .r 2  $ $QY / 5 5d6M6M N  # #A'7'7$7 8 > >t?U?U Vr3c|jdg|_i|_|j|jj yy)z4Reset all relevant states for a new generation loop.TrAN)rZrRrSr] synchronizer_s r4resetzContinuousBatchingIOs.resetsF ""d"3!#,.)    *    + + - +r3c2|j|jfS)z:Get the cumulative sequence lengths for the current batch.)ryr{rs r4get_cumulative_seqlensz,ContinuousBatchingIOs.get_cumulative_seqlens(s(($*C*CCCr3rrzprev_output_idscyNr2)r_rrzrs r4carry_over_tokensz'ContinuousBatchingIOs.carry_over_tokens,s r3cR|j|jjyyr)r]rrs r4retrieve_device_outputsz-ContinuousBatchingIOs.retrieve_device_outputs1s%    *    + + - +r3c&|jdd|jfj}|jrJ|jdd|jfj t j j}nd}|j||fS)Nrr re)r~rLtolistrIviewr*float32rR)r_ new_tokenslogprobss r4prepare_batch_updatez*ContinuousBatchingIOs.prepare_batch_update5s__Q(C$*C*C(C%CDKKM   q*ED,E,E*E'EFKKRWR_R_K`ggiHH%%z8;;r3rRlogits_processorsuse_decode_fast_pathrJrK use_paddingc  |s td|xr|jjdkD|_||_|jrdn||_t ||_|jt|jjDcgc]}dc}|_ t|jjDcgc]}dc}|_ g|_i|_g}g} dg} g} |j j#D cic]} | dg} } t|jjDcgc]}g}}|j dk(rdn,t|jjDcgc]}gc}}t%|D]\}}|j&}|j(}|j*}|jj-||}|xj(|z c_|j/|j0| j/t|||z| j3| d|zt5|j6||_|j9D]H\} }| | j3| | d|zt5|j:| ||j:| <J|jr:|jj=|j>|||jdd|fn)|jjA|j>|||||jBr?| j3| ddz tDg|_| d|j|j><|jj3||jG||jH|jJd| d|_&|jNE| j9D]2\} }tQ|jN| | || dk(r |jRnd4|rG|jU| }||jz }| j/tW|jL|nd}tYtZj\tZj^|j` }|||jbdt ||| |jddt | || |jfdt | || |jhdt | | j9D]U\} }|d}||jj| <|j/tW|||||j | dt |W|jstYtZj\tZjl|j` }t%|D]?\}}|||jn|dt |f<t ||j|<A|Nt%|D]?\}}|||jp|dt |f<t ||j|<Ayyycc}wcc}wcc} wcc}wcc}w) aPrepare tensors and metadata for the next model forward pass, using the given requests as data. This method: 1. Resets the static tensors from the previous batch 2. Iterates through requests to accumulate input_ids, position_ids, and sequence lengths 3. Extends read/write indices for cache management 4. Builds attention masks if needed (for eager/SDPA implementations) 5. Converts accumulated lists to tensors and copies them to static storage This method also modifies the `position_offset` attribute of each request to track progress and adds a temporary token at the end of the requests for which there will a new token. zNo requests in batchrNrr )rR arg_storagerm)rryr{r@rrg)9 ValueErrorr#numelrQrJrKrrrLrZrMr"rNrOrPrRrSr{r enumeratestateposition_offset query_length get_seqlens_kextendtokens_to_processappendmaxritemsrfill_block_table request_idextend_read_and_write_indices has_new_tokenrprepare_tensor_argsrvr8rrrr@_get_num_sequencesrrr*tensorrur;rrryr!rrrr)r_rRrrrJrKrr`rrryr!rr{rr i future_stater past_lengthr seqlens_klayer_type_seqlen_klayer_type_seqlens_knum_sequences_in_next_batchfake_sequences to_tensorrto_index_tensorgroup_write_indicesgroup_read_indicess r4prepare_batch_tensorsz+ContinuousBatchingIOs.prepare_batch_tensors?s(*!34 4 4T8H8H8N8N8PST8T( $ 4 41+$'(9$:! ""$,11F1F+GHaH,1$**2G2G,H Iq I!#,.)  !sBFB[B[B`B`BbcJ QCcc#()>)>#?@ar@ @!--2TU4::K`K`Ea8b8b  )):;# 8OA| &&E//K'44L 00lKI  ! !\ 1 !   U44 5   k;3M N O ' '(&BC+7.'FTUWFX11%2B2BC  " " ) ), 7G# 8L --///0B0B0DE . 326    *4H4N4N4P 0 0$#'#6#6z#B)=)=:DH[:[4#6#6ab   *.*A*Ak*A*Z '84;T;TTN ' 't/B/BN(S TNELL DKKP ,5Y+?'Y(1:<1H-C -.AJK_A`!!"=C(<$=>5>~5N1c.121E0J0J0L q ,J,1"5N."&"9"9$:L:L:NP^Q^P^:^"_#'#6#6#>4B**?C?S?S((N]N):;Y] $  &D ''D,F,F'+&7&7&<&<&>%" D,=,=j,I4::K_K_ae ff%!'+&:&:@Q@Q~tzz,,- VA)4v$:O:OPQ:R  = ! ( ()A)A!EVFVEVBV)W X1$"#-8'd>R>RST>U < ' '(?(?CSOCS@S(T U V&*%>%>%D%D%F m !J 2;O2PF? #J /6:6J6JPTPaPablPmF> ": ."".#.D4G4G 4S7;7J7J:7VWZ\c]c\cekfkekWk7l'(4  m t((--/ 0A 5&,_&=&E&E&G&JF? #%+N%;%C%C%Ea%HF> """.+12B+C+K+K+Ma+P'( G%s6McH|j|j|jfS)a.Returns the tensors used inside the generation step that are not inputs to the model forward pass. In synchronous batching, there is no carry over, so the only tensor that will be used is output_ids, but we still return 3 tensors to have the same interface as when using async batching.)rzr~rs r4 get_cb_kwargsz#ContinuousBatchingIOs.get_cb_kwargss""DOOT__DDr3.c|jr |jfS|j|jg|jj Sr)rQrJrKrvaluesrs r4_get_graph_keyz$ContinuousBatchingIOs._get_graph_keysE   %%' '!!4#3#3Qd6G6G6N6N6PQQr3prefixc|j}|jj|}|tj|d||S)NzCreating graph for key = )rrT get_graphrinfo)r_rkeygraphs r4rzContinuousBatchingIOs.get_graph%sF!!# %%c* = KK6(" ? r3rc|j}|jj||tjd|y)NzSetting graph for key = )rrT set_graphrr)r_rrs r4rzContinuousBatchingIOs.set_graph-s8!!# c5) /12r3Fr>N))+r&r'r(r)r8r/r,r rrr*r;rer rarYrCr\r1rno_gradrZrtupler+r-r.rrrr0rfloatrrrrrrr CUDAGraphrrr2r3r4r6r6Fs M3*g"*g!*g%= *g  *g [[ *g?*g *gXW 5ZW _cW v_dw,w6;jj6G6GwW[w w@U]]_(W(W(W(WT.DellDellARNc t|||tjd|||_t|||||||_tj j rtj jnd|_tj j rtj jnd|_ tj j r$tj j|_ yd|_ y)Nrcr"r9r:r;r<r=) r6r*r;host_io device_iorC is_availableEventh2d_over compute_overd2h_over)r_r"r9r:r;r<r=s r4razHostDeviceIOPair.__init__4s-'A<<&#+   /'A#+  /4jj.E.E.G ((*T 27**2I2I2KEJJ,,.QU.3jj.E.E.G ((* T r3c|jj|jj|j|j|j fD]}||j yr)rrrrrrr)r_events r4rzHostDeviceIOPair.resetSsX  mmT%6%6 F $E !!# $r3rcT|jj|j|dy)NT)rr)rrr)r_rs r4transfer_inputs_h2dz$HostDeviceIOPair.transfer_inputs_h2dZs  %%dnnVRV%Wr3c|tjj|n t}|5|jj j |jj ddddy#1swYyxYw)NTr)r*rCrrrr~rr)r_rrs r4transfer_outputs_d2hz%HostDeviceIOPair.transfer_outputs_d2h]sc4:4Fuzz((0KM  X LL # # ) )$..*C*CRV ) W X X Xs $X%***;*;XXX5::+<+│1->│ ¦ │2->│ ¦ │3->│ ¦ ¦ │ └───┼───┴───────────┬─────────────┴─┬─┼───────────┴───┼───────────────┐ ¦ GPU │ │ COMPUTE 0 │ COMPUTE 1 │█│ COMPUTE 2 │ COMPUTE 3 │ ¦ │ └───────────────┼───┬───────────┼─┴─┬─────────────┼───┬───────────┼───┤ D2H │ │0<-│ │1<-│ │2<-│ │3<-│ │ └───┘ └───┘ └───┘ └───┘ with: - CPU: actions happening on the CPU (host-side) - GPU: actions happening on the GPU (device-side) - H2D: host to device transfer - D2H: device to host transfer and: - PR N: preparation of batch N - ->N: host to device transfer of batch N - COMPUTE N: compute step for batch N - <-N: device to host transfer of batch N - UP N: update of batch N You can see that the GPU is almost always busy, except where the █ is. Proper ordering of steps is ensured through the use of CUDA events and streams. r"r9r:r;r<r=r>Nc tjjs*tdtjjd|_t dDcgc]}t ||||||c}|_tjj||_ tjj||_ tjj||_ d|jdj_ d|jdj_ d|jdj_ d|jdj_ |j|_ycc}w)Nz>Async batching requires CUDA, but torch.cuda.is_available() = rrhrrDr )r*rCr RuntimeError current_pairrMrio_pairsr\ h2d_stream d2h_streamr]rrror^s r4raz#ContinuousBatchingAsyncIOs.__init__s3zz&&(!`EJJD[D[D]Cabc c1X   +E' /      **++6+:**++6+:#jj//v/>26 a  /48 a""126 a  /48 a""1 % 6 6+  sE6cd|j|jjjSr)rrrrrs r4rz1ContinuousBatchingAsyncIOs.get_cumulative_seqlenss&}}T../77NNPPr3rRrrrJrKrc|j|j}|jj|||||||jjj |j yr)rrrrrzrinfer_carry_over_ids)r_rRrrrJrKrio_pairs r4rz0ContinuousBatchingAsyncIOs.prepare_batch_tensorss^-- 1 12-- 02F Vacn  &&,,T-F-F-HIr3c|j|jjj}|jd|jz jj}t |j Dcgc]}d}}t |jD]\}}|j|}||||< tj|tjScc}w)aInfers the ids of the tokens to carry over from batch N to batch N+1. In asynchronous batching mode, we can schedule a request for batch N+1 without knowing the token predicted for that request in batch N. For that reason, we might need to carry over tokens just predicted in batch N before launching the forward pass of batch N+1. This method computes the ids of the tokens to carry over.r rr) rrrrSrMrorrgetr*rru)r_!next_req_id_to_new_token_position!prev_req_id_to_new_token_positionr`rzrreq_idnew_token_positions r4rz/ContinuousBatchingAsyncIOs.infer_carry_over_idss -1MM$:K:K,L,T,T,q,q),0MM!d>O>O:O,P,X,X,u,u)&+D,A,A&BC"CC ##D#I#I#KL 7IAv!B!F!Fv!N !-5612 7||N%++>>Ds5 Cc6|j|j}|j|j|jj |j |j j|j |jj|S)Nr) rrrr record_eventrr] wait_eventrr)r_rrs r4rz+ContinuousBatchingAsyncIOs.get_model_kwargsst-- 1 12##DOO4 $$W%5%56 &&w'7'78  11k1JJr3c|j|j}|jd|jz }|jj|jj|jjfS)aReturns the tensors used inside the generation step that are not inputs to the model forward pass. Those tensors could be retrieved using this object, but it would trigger a recompile if using torch.compile. They are: - output_ids: the output ids of the current batch - prev_output_ids: the output ids of the previous batch, required to carry over outputs tokens of the previous batch to the input tokens of the next batch. - carry_over_ids: a mask representing how to carry over tokens. r )rrrrzr~)r_r previous_pairs r4rz(ContinuousBatchingAsyncIOs.get_cb_kwargssh}}T%6%67  a$*;*;&;<  " " 1 1  # # . .  " " - -  r3rrzrc|d|f}|dk7j}|d|jd}|d|jd}||z|dd|z zz|d<y)aAs explained in the infer_carry_over_ids method, we might need to carry over tokens just predicted in batch N before launching the forward pass of batch N+1. This method performs the carry over, and is recorded in CUDA graphs if they are enabled.rrNr )r/rn)r_rrzrcarried_over_idscarried_over_masks r4rz,ContinuousBatchingAsyncIOs.carry_over_tokenssy+1n+<=+r1668+,?innQ.?@-.A q0AB'*;;ilaRcNc>dd ! r3c\|j|jjjSr)rrrr~rs r4r~z%ContinuousBatchingAsyncIOs.output_ids s%}}T../99DDDr3cd|jd}|j|jjj|S)Nz(IO ))r)rrrr)r_rs r4rz$ContinuousBatchingAsyncIOs.get_graphs@))*!,}}T../99CC6CRRr3rch|j|jjj|yr)rrrr)r_rs r4rz$ContinuousBatchingAsyncIOs.set_graphs% d''(22< ""7#7#78$$T__5 $$W%5%56 r3c,d|jz |_y)zeSwitch to the other IO pair so the next batch reads from / writes into a fresh set of static buffers.r N)rrs r4r+z(ContinuousBatchingAsyncIOs.swap_io_pairs%s 1 11r3c|j|j}|jj|jj Sr)rrrrrrr,s r4rz/ContinuousBatchingAsyncIOs.prepare_batch_update*s;-- 1 12$$&3355r3cd|_|jD]}|j|jj |j j |j j y)zWReset all state for a new generation session. Used in persistent mode between sessions.rN)rrrrrrr]r,s r4rz ContinuousBatchingAsyncIOs.reset/sX}} G MMO  ##% ##% '')r3rr)'r&r'r(r)r rrr*r;rer rarr+r-r.rr0rr1r/rrrrrrpropertyr~rCrrrrQrr+rrrr2r3r4r r cs+Z#7"#7!#7%= #7  #7 [[ #7?#7 #7LQellDellARr:sd# =P*&G>>jj$$Nj3j3Z-X-X`S*S*r3