o i@s0ddlZddlZddlZddlZddlZddlmZmZddlm Z ddl m Z m Z ddl mZddlmZddlmZmZddlZddlmZddlmZmZmZdd lmZd d lmZd d lmZd d l m!Z!d dl"m#Z#m$Z$m%Z%GdddeZ&e'e(Z)e)*ej+e GdddZ,e GdddZ-e$GdddeZ.GdddeZ/e$Gddde/Z0e$Gddde/Z1e%ddej2ddfd ej3d!ed"e!d#e4e4e5d$ej6d%e7d&ee5fd'd(Z8e Gd)d*d*Z9e%d+d,Z:e$Gd-d.d.Z;e0e1d/ZdS)4N)ABCabstractmethod)deque) dataclassfield)Enum)partial)OptionalUnion)profilescheduletensorboard_trace_handler)tqdm)Cache)PretrainedConfig)GenerationConfig)ContinuousBatchProcessorMetrics attach_tracertracedc@s,eZdZdZdZdZdZdZdZdZ dZ d S) RequestStatusz5Status of a generation request through its lifecycle.pending prefillingprefilling_splitsplit_pending_remainderdecodingfinishedfailedN) __name__ __module__ __qualname____doc__PENDING PREFILLINGPREFILLING_SPLITSPLIT_PENDING_REMAINDERDECODINGFINISHEDFAILEDr)r)n/home/ubuntu/maya3_transcribe/venv/lib/python3.10/site-packages/transformers/generation/continuous_batching.pyr'src@seZdZUdZeed<eedZee ed<eedZ ee ed<eedZ ee ed<dZ eeed<ejZeed <eejdZe ed <dS) GenerationOutputaTracks the output of a generation request. Attributes: request_id (str): The ID of the generation request. prompt_ids (list[int]): The IDs of the prompt tokens. generated_tokens (list[int]): The generated tokens. logprobs (list[float]): The log probabilities of the generated tokens. error (Optional[str]): Any error message associated with the request. When None, the request was successful. request_iddefault_factory prompt_idsgenerated_tokenslogprobsNerrorstatus created_time)rrr r!str__annotations__rlistr/intr0r1floatr2r rr"r3timer4r)r)r)r*r+8s  r+c@seZdZUdZeed<dZeee ed<dZ eee ed<e edZ ee ed<e edZ ee ed<e edZee ed <d Ze ed <ejZeed <d Ze ed<dZe ed<e ejdZeed<dZeeed<de fddZde fddZede defddZddZddZdS) RequestStatezTracks the state of a generation request through its lifecycle. Attributes: status (RequestStatus): can be one of PENDING, PREFILLING, PREFILLING_SPLIT, SPLIT_PENDING_REMAINDER, DECODING, FINISHED, FAILED r,Nr/full_prompt_idsr-remaining_prompt_idsstatic_outputsallocated_blocksrposition_offsetr3max_new_tokens eos_token_idr4r2returncCs|jS)zCGet the current length of the sequence (prompt + generated tokens).)r@selfr)r)r* current_lendszRequestState.current_lencC t|jS)z*Get the number of tokens generated so far.)lenr>rFr)r)r* generated_lenh zRequestState.generated_lentoken_idcCs`|jtjkrdS||jko|jdk}||jk}|r|r$|j|g|s(|r.tj|_dSdS)zUpdate the request with a newly generated token and check for completion. Args: token_id: The token ID to add to the output sequence Returns: bool: True if the request is now complete, False otherwise FrCT) r3rr&rDrKrBr>extendr')rGrMis_eos is_max_lenr)r)r*update_with_tokenls zRequestState.update_with_tokencCs\d|jd|jd|dt|jdt|jd|jdt|jd|jd |j d S) NzRequestState( request_id=z , status=z, out_tokens=z, query_length=z, remaining_tokens=z, kv_length=z full_prompt_lenght=z, allocated_blocks=z, generated_tokens=z )) r,r3rKrJr/r=r@r<r?r>rFr)r)r*__repr__s\zRequestState.__repr__cCst|j|j|j|jg|jdS)z7Convert the request state to a GenerationOutput object.)r,r/r3r0r1r2)r+r,r<r3r>r2rFr)r)r*to_generation_outputsz!RequestState.to_generation_output) rrr r!r5r6r/r r7r8r<rr=r>r?r@rr"r3rBrDr:r4r9r2rHrKrboolrQrRrSr)r)r)r*r;Ms(     r;c@s eZdZejddfdededejdejde e e e e eje ffde eee ddfd d Zed e d e dee fd dZed e ddfddZde fddZd e dee fddZededee dee fddZedejdejde deejejffddZdS)PagedAttentionCacheNconfiggeneration_configdevicedtypelayer_device_mapinitial_prompt_shapesrEc Cstt|dddur |jn|j|_t|dr|jn|j|j|_|j|_t|dd}t|dd}|dus6|durTtdt ||||pBg|dd\}}td |d |||_ ||_ |j||j |jf|_ ||_ ||_g|_g|_t|jD]8} |dur|| n|} tj|j |j | d } tj|j |j | d } tj| tj| |j| |j| qutt||_i|_dS) aInitialize a paged attention cache for efficient memory usage. Args: config: Model configuration generation_config: Generation configuration containing cache parameters device: Device for the cache tensors dtype: Data type for the cache tensors layer_device_map: Optional mapping of layer indices to devices initial_prompt_shapes: Optional sample prompts to help calculate optimal cache size num_key_value_headsNhead_dim num_blocks block_sizez,Calculating optimal block size and number...)median_prefill_lengthzUsing calculated num_blocks=z , block_size=rYrX)getattrnum_attention_headsr\hasattrr] hidden_sizenum_hidden_layersloggerinfocompute_optimal_blocksr_r^ cache_shaperYrX key_cache value_cacherangetorchzeros_dynamomark_static_addressappendr _free_blocks _block_tables) rGrVrWrXrYrZr[r^r_idx layer_devicenew_layer_key_cachenew_layer_value_cacher)r)r*__init__s@        zPagedAttentionCache.__init__n_blocksr,cCs\t|j|kr dSg}t|D] }||jq||jvr$g|j|<|j|||S)z*Allocates n_blocks for a given request_id.F)rJrtrnrspopleftrurN)rGr{r, allocated_r)r)r*allocate_blockss   z#PagedAttentionCache.allocate_blockscCs:||jvr|j|}|j|dStd|dS)z.Frees all blocks associated with a request_id.z6Attempted to free blocks for non-existent request_id: N)rupoprtrNrhwarning)rGr,blocks_to_freer)r)r* free_blockss  zPagedAttentionCache.free_blockscCrI)z,Returns the number of free blocks available.)rJrtrFr)r)r*get_num_free_blocksrLz'PagedAttentionCache.get_num_free_blockscCs|j|gS)z&Returns the block table for a request.)rugetrGr,r)r)r*get_block_tablez#PagedAttentionCache.get_block_tablestatelogical_indicesc Cs|j}|j|}|std||j}g}|D],}||}||} |t|kr6td|d|d|||} | || } || q|S)a Maps logical sequence indices to physical cache indices using the block table, using PyTorch. Args: request_id: The request ID. logical_indices: A list of logical indices. Returns: A list of physical indices. Raises: ValueError: If no block table is found for the request ID. IndexError: If a logical index maps to a block index that is out of bounds. z!No block table found for request zLogical index z maps to block index z$ which is out of bounds for request )r,rur ValueErrorr_rJ IndexErrorrs) rGrrr, block_tabler_physical_indicesrv block_idx block_offsetphysical_block_numphysical_indexr)r)r*_get_physical_indicess&    z)PagedAttentionCache._get_physical_indices key_states value_states layer_idxc Ks|j|j}|j||j||j}|j||j||j} |d|dd|ddf<|d| dd|ddf<|ddd|ddf| ddd|ddffSNr)r^r_rlviewr\r]rm) rGrrr read_index write_indexkwargs total_slots k_cache_flat v_cache_flatr)r)r*update s 0zPagedAttentionCache.update)rrr rofloat16rrrXrYr dictr8r r5r7rzrrrrrr;rTensortuplerr)r)r)r*rUsJ @ 'rUc@seZdZdZddedefddZedefdd Z ed e d e efd d Z e d efddZeddedefddZe ded e e fddZdS) Schedulerz Abstract base class for scheduling requests in the continuous batch processor. It is expected that cache allocation and scheduling logic will be implemented in subclasses. Fcacheretain_cache_on_finishcCs$i|_i|_t|_||_||_dSN)active_requestswaiting_requestsrwaiting_requests_orderrr)rGrrr)r)r*rz9s  zScheduler.__init__rcCdSz"Add a request to the waiting list.Nr))rGrr)r)r*add_waiting_request@zScheduler.add_waiting_request token_budgetrEcCsdSrr))rGrr)r)r*schedule_batchEszScheduler.schedule_batchcCs |jp|jS)z2Check if there are requests ready to be processed.)rrrFr)r)r*has_pending_requestsIs zScheduler.has_pending_requestsTr,evict_from_cachecCr)z:Finish processing a request and free its allocated blocks.Nr)rGr,rr)r)r*finish_requestNrzScheduler.finish_requestcCs||jvr |j|jSgSr)rr>rr)r)r*!get_active_request_static_outputsSs  z+Scheduler.get_active_request_static_outputsNFT)rrr r!rUrTrzrr;rr8r7rrrr5rrr)r)r)r*r3src@eZdZededefddZedddededeefd d Z edefd d Z eded e efddZ eddede fddZdS) FIFOSchedulerrlen_next_tokenscCt|}t|j|jj|}||kst|jdkr8||d|jjd}|j||j}|s2dS|j|dSNrFTrHrJr?rr_rr,rNrGrrrH occupancy blocks_neededr}r)r)r*_allocate_blocks_if_needed\ z(FIFOScheduler._allocate_blocks_if_neededprepare_request span_namer"request_ids_to_remove_from_waitingcC|jtjkr |jn|j}t||kr?|jtjkr*||j|j<tj |_| |jdS|jtjkr=tj |_|j|_g|_dSdS|jtjkrV||j|j<tj |_| |jn |jtjkr`tj |_||d|_|d||_dSz6Prepare a request for processing in the current batch.N r3rr%r=r/rJr"rr,r#addr$rGrrrrequest_tokensr)r)r*_prepare_request_for_processingj(        z-FIFOScheduler._prepare_request_for_processingcCd|jr#|j|jvr#|j|j}|jt|jd|_|j|_|j|_||j |j<|j |jdSr rr,rrr/rJr<r?r@rrrsrGr old_stater)r)r*r z!FIFOScheduler.add_waiting_requestrEc s,g}g}gjD]}|jtjkr|||jtjkr#||q jD] }|j|q'||}t |D]K} ||t |j } |t |j s]t jjdkr\n+q;tdtffdd }||||8}tdtffdd } | ||dkrnq;tfddjD_S) Nrrc|dSrrsrscheduled_requestsr)r*_add_to_scheduled_requestsrz@FIFOScheduler.schedule_batch.._add_to_scheduled_requestsc*|j}|jvrj|=|dSdSrr,rrrreq_idrrGr)r*_remove_from_waiting_requests  zCFIFOScheduler.schedule_batch.._remove_from_waiting_requestscg|]}|vr|qSr)r).0rrr)r* z0FIFOScheduler.schedule_batch..)rvaluesr3rr&rsr%rrsetrrJr/rrrtrr;r rGrpriority_statessecond_priority_statesrr candidates request_lenrrr)rrrGr*rsH       zFIFOScheduler.schedule_batchTr,rcC.|r|j|||jvr|j|=dSdSdSrrrrrr)r)r*r    zFIFOScheduler.finish_requestNrrrr rr;r8rrr5rrr7rrTrr)r)r)r*rZ"   6rc@r)PrefillFirstSchedulerrrcCrrrrr)r)r*rrz0PrefillFirstScheduler._allocate_blocks_if_neededrrrrcCrrrrr)r)r*rrz5PrefillFirstScheduler._prepare_request_for_processingcCrrrrr)r)r*rrz)PrefillFirstScheduler.add_waiting_requestrEc s.g}g}gjD]}|jtjkr||q |jtjkr$||q jD] }|j|q(||}t |D]K} ||t |j } |t |j s^t jjdkr]n+q._add_to_scheduled_requestscrrrrrr)r*r.rzKPrefillFirstScheduler.schedule_batch.._remove_from_waiting_requestscrr)r)rrr)r*r;rz8PrefillFirstScheduler.schedule_batch..)rrr3rr%rsr&rrrrrJr/rrrtrr;rrr)rr*r sH       z$PrefillFirstScheduler.schedule_batchTr,rcCrrrrr)r)r*r@rz$PrefillFirstScheduler.finish_requestNrrr)r)r)r*rrrT standaloneg?rXrVrWinputsrY safety_marginrac Cst|d|j|j}t|d|j}t|dd} |jdkr9tj|} | j} tj|} tj |} | t | | }n|jdkrEt ddSt d |jd d St ||}|d krbt d dStjg|d}d|||| }t|dd}|dur|rdd|D}|rt t|nd}n|durd}||}d}||}t dt ||}||}t |||d}d|d>}t d||}t d|d|d|||d|d t |t |fS)aCalculate optimal number and size of blocks for the KV cache. Args: device: The device where the model runs config: The model configuration generation_config: The generation configuration inputs: Sample input sequences to estimate memory requirements dtype: Data type for cache tensors safety_margin: Fraction of available memory to use median_prefill_length: Override for median prefill length calculation Returns: Tuple of (num_blocks, block_size) r]r\rg(cudampszBMPS memory estimation is approximate. Using conservative defaults.)zUnsupported device type z/ for optimal block calculation. Using defaults.) rz9Not enough available memory. Using minimum configuration.)r)rYrrBrANcSsg|]}|rt|qSr))rJ)rseqr)r)r*rrz*compute_optimal_blocks..@rzOptimal cache: z blocks of size z (can handle ~z sequences of length ))rcrfrdtyperorget_device_properties total_memorymemory_allocatedmemory_reservedmaxrhrr8tensor element_size statisticsmedian bit_lengthri)rXrVrWrrYrrar] num_kv_headsrgdevice_propertiesrallocated_memoryreserved_memoryavailable_memory dtype_sizememory_per_tokentokens_to_generatenon_empty_inputs seq_lengthMIN_BLOCK_SIZEper_sequence_memorymax_concurrent_sequences total_tokensinitial_block_sizer_r^r)r)r*rjHsT            rjc@seZdZUejed<ejed<ejed<ejed<ejed<eed<eed<ejed<ejed <ejed <eee efed <e ed <d Z e ed<dS)PagedAttentionArgs input_idsattention_mask position_idscumulative_seqlens_qcumulative_seqlens_k max_seqlen_q max_seqlen_krrlogits_indices block_tablesrF use_cacheN) rrr rorr6r8rr5r7rUr*rTr)r)r)r*r s        r cCs`|dd|ddk}|dd|ddk}t||}|d|d}|d|d}|d}|d}tj||jd}tj||jd}tj||dddd} tj||dddd} | dddf| dddfk} |dd|dddk|dd} tj|| dddddf| k} tjtj||| jddd}| | |@d| S)NrrC)rXT)rightFdiagonal) minsumroarangerX bucketizetriuonesrT masked_fill_)r$r% valid_docs_q valid_docs_knum_valid_docstotal_qtotal_k q_indices k_indices q_doc_ids k_doc_idsdoc_mask is_causal apply_causal causal_maskr)r)r*create_document_masks" *  rBc@s.eZdZ  d3dedededejdejdej de j d e j d e d ed efd dZeddddZee ddZdefddZddZeddddZeddZedefdd Zed!d"Zed#d$Zed%d&Zeded'efd(d)Zed*d+Z edefd,d-Z!ed.d/Z"ed0d1Z#d2S)4ContinuousBatchProcessorFrrVrW input_queue output_queue stop_event model_device model_dtype scheduler streamingmanual_evictionc Csh||_||_||_||_||_||_||_||_| |_| |_ | |_ g|_ | t |j|_|dS)aInitialize the continuous batch processor. Args: cache: The paged attention cache to use 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 streaming: Whether to stream tokens as they're generated N)rrVrWrDrErFrGrHrIrJrKrequests_in_batch_configure_batch_parametersrmax_batch_tokensmetricssetup_static_tensors) rGrrVrWrDrErFrGrHrIrJrKr)r)r*rzs  z!ContinuousBatchProcessor.__init__TrcCs|j}|jj|jj}tj|jd}||_tjd|ffi||_ tjd|ffi||_ tjdd||f|j |jd|_ tj|dffi||_ tj|dffi||_tj|ffi||_tj|ffi||_tj|fdfi||_d|_d|_tjd|fdfi||_dS)NrbrrCr)rNrr^r_roint32rGtensor_metadatarpr!r#rHr"r$r%rrfullr(r&r' output_ids)rGTmax_token_budgetrRr)r)r*rPs"z-ContinuousBatchProcessor.setup_static_tensorscCs||j|j|jt|jj|j |j |j d|j d|j dd|_d|_|jdS)z(Reset static tensors for the next batch.rCrN)r!zero_r#r"fill_rofinforHr.r$r%rrr(r&r'rTrFr)r)r*reset_static_tensorss       z-ContinuousBatchProcessor.reset_static_tensorsrEcCs:|j|j|j|j|j|j|j|j|j|j |j j |j dd S)z2Get model keyword arguments for the current batch.F) r!r#r"r$r%rrr(r&r'r)rr*) r!r#r"r$r%rrr(r&r'rrurFr)r)r*get_model_kwargs(sz)ContinuousBatchProcessor.get_model_kwargsc Cs6d|jd|jd|jjd|jjd |S)Nz%ContinuousBatchProcessor(input_queue=z, output_queue=z, active_requests=z, waiting_requests=r)rDrErIrrr[rRrFr)r)r*rR;s( z!ContinuousBatchProcessor.__repr__cCsn|jj|jj}t|jdd}|dur||_nt|jdd|_t|d|j}td||_t|jdd|_dS)z>Set up batch processing parameters based on generation config.rNNmax_position_embeddingsrrr) rr^r_rcrWrNmax_context_lenr.r )rGtotal_cache_tokensuser_batch_tokensrecommended_batch_sizer)r)r*rMAs z4ContinuousBatchProcessor._configure_batch_parametersc Cs|jsVz|j}|durWq|j|Wn6tjy#YdStyN}z tj d|ddt d}|durD| ||WYd}~nd}~ww|jrdSdS)z?Pull new requests from the input queue and add to waiting list.NzError processing new request: Texc_infor) rDempty get_nowaitrIrqueueEmpty Exceptionrhr2localsr_handle_request_error)rGrer)r)r*_get_new_requestsTs    z*ContinuousBatchProcessor._get_new_requestsrcCs\tj|_t||_t|jtr|j|j|_ ng|_ |j |j |j|j |dS)z(Handle general request processing error.N)rr(r3r5r2 isinstancer,rIrr>rOrecord_request_completionr4rEputrS)rGr2rr)r)r*rifs  z.ContinuousBatchProcessor._handle_request_errorcCs||js dS|jt|jjt|jj|j|j |_ |j s'dS| g}g}g}g}dg}dg}g}|j |j |j D]u}|j } || |j} t| } | | } tt| } | | d}|j|| }|| d}||||||||d| ||d| t|jdkr||ddt|j| |_t|j| |_|j| 7_qEtdt|j dt|jjdt|jjd|dd |dd |j |||||||||j|jdS) z=Prepare tensors and metadata for the next model forward pass.NrrCrz Scheduled: z , Waiting: z , Active: z . cum Q: z . cum KV: z, free blocks: )rkrIrrOrecord_queue_metricsrJrrrrNrLrZrecord_batch_metricsr/rNr@r7rnrrrsr=r r&r'rhrr_build_tensorsrecord_kv_cache_memory_metrics)rGr#r!rrr$r%r(rnext_input_ids past_length query_length key_length cache_indexpositions_to_add read_indices write_indicesr)r)r*prepare_next_batchusb        J z+ContinuousBatchProcessor.prepare_next_batchcCsttjfi|j}|||jdddt|f<|||jdddt|f<|||jdt|<|||jdt|<|||j dt|<|||j dt|<|||j dt|<t |j j} |jjdkrtt|dD]t} || d|| || d|| kr|| d|| dkr|| d|| d|| d} | || } nd} t|| || d} t|| || d} tjtj|jd| | fj| |j |jd| d}||jd| | f<qtdSdS)Npaged_attentionr.rbr,)rror rRr!rJr#rrr$r%r(rYrHr.rV_attn_implementationrnslicer2rSr"shaperG)rGr!r#rrr$r%r( to_tensor min_valueir- query_range key_rangemaskr)r)r*rqsB  " z'ContinuousBatchProcessor._build_tensorscCs|jdSr)rTtolistrFr)r)r*_syncrzContinuousBatchProcessor._synctokencCsD|jr||_|j|dS|jtjkr |j|dSdS)zCSend output to the queue based on streaming mode and request state.N)rJ next_tokenrErnrSr3rr')rGrrr)r)r*_maybe_send_outputs  z+ContinuousBatchProcessor._maybe_send_outputcCs|}g}t|jD]U\}}|j}t|jdkrV|j|j|jt j |_ ||j |}|g|_ ||rO|j|j|j|jj|j|j d|||||q |j t jkr`t j|_ q dS)z0Update request states based on generated tokens.r)rN)r enumeraterLr,rJr=rOrecord_ttft_metricr4rr&r3r(r/rQrmrIrrKrsrr$r%)rG out_tokensfinished_request_idsrrrrr)r)r* update_batchs$   z%ContinuousBatchProcessor.update_batchcCs |jS)z2Check if there are any active or waiting requests.)rIrrFr)r)r*rs z-ContinuousBatchProcessor.has_pending_requestscCs.|j}|D]}||||j|jqdS)z&Handle errors during batch processing.N)rLrirIrr,)rGr2 failed_reqsreqr)r)r*handle_batch_error s  z+ContinuousBatchProcessor.handle_batch_errorcCsl|jjD]}||||j|jqt|jjD]}|jj |}|||q|jj dS)zFail all active requests with the given error. Args: error: The error to report in the failure message N) rIrrrirr,r7rkeysrrclear)rGr2rrr)r)r*fail_all_requestss z*ContinuousBatchProcessor.fail_all_requestsN)FF)$rrr rUrrreQueue threadingEventrorXrYrrTrzrrPno_gradrZr r[rRrMrkr;rir{rqrr8rrrrrr)r)r)r*rCsn     0     > .   rC)fifo prefill_firstc @seZdZdZ   d>dededefdd Zed d Zd d Z d?dede e fddZ d@de e fddZ dAdeede ede edefddZdeeefddZd@de efddZd d!Zed"d#Zed$efd%d&Zed'd(d)d*Zed+d(d,d-Zed.d(d$efd/d0Zd1d2Zed3d(dBd$ed4efd5d6Zed7d(d8d9Zed$e efd:d;Zedefds         H'vv  Y UD