o Ti(@sddlZddlZddlZddlmZmZddlZddlmZ ddl m Z ddl m Z ddlmZddlmZddlmZmZmZdd lmZmZdd lmZmZdd lmZdd lmZd Z GdddZ!dS)N)IterableTuple)get_accelerator)init_distributed)InferenceV2Policy)inference_logger)DSStateManagerRaggedBatchWrapperPlaceholderSequenceDescriptor)SchedulingErrorSchedulingResult)make_param_filenamemake_metadata_filename)DSInferenceModelBase)RaggedInferenceEngineConfigzmodel-forward-inferencec @seZdZUeed< eed< eed< edej fddZ ede fddZ defd d Z d ed edd fddZddZ d'dee deej dedej fddZde de dee ej ffddZdee dee defddZde de fd d!Zde dd fd"d#Zd$edd fd%d&Zd S)(InferenceEngineV2_config_model_state_managerreturncC|jjS)z Number of free KV blocks. This is a tensor of shape [n_kv_cache_groups] where each element is the number of free blocks in the corresponding KV cache group. )r free_blocksselfrT/home/ubuntu/.local/lib/python3.10/site-packages/deepspeed/inference/v2/engine_v2.pyr/szInferenceEngineV2.free_blockscCr)z, Number of KV cache groups. )rn_kv_cache_groupsrrrrr7sz#InferenceEngineV2.n_kv_cache_groupscCs|jS)z+ The model implementation. )rrrrrmodel>szInferenceEngineV2.modelpolicy engine_configNcCs~||_||_||_td|j|j|j|_tdt|jj |_ t |jj |j |jd|_ |j|j dS)ab Create the Inference V2 engine. Arguments: policy (InferenceV2Policy): Policy for the model implementation. This policy object will be used to build the model and load the checkpoint associated with it. engine_config (RaggedInferenceEngineConfig): Configuration for the inference engine. zBuilding model...z Model built.) base_mp_groupN)r_policy_initialize_tp_group_base_mp_grouprinfo build_modelrr state_manager_batchr kv_cache_configrset_state_manager)rrr rrr__init__Ds    zInferenceEngineV2.__init__cCsVtttdd}t|||jjjkrt dt t |jjj}t j |dS)z@ Implementation of our TP group initialization. LOCAL_RANKrzILocal rank is greater than TP size, ensure that the TP config is correct.)ranks)rintosgetenvr set_devicertensor_paralleltp_size RuntimeErrorlistrangedist new_group)r local_rankr-rrrr#]s  z&InferenceEngineV2._initialize_tp_groupT batch_uids batch_tokens do_checksc Cs|rdd|D}|||}|tjkrt||jt||D]#\}}|j|}|j || | | |jj |||dq"|j|j |j|j |j} | jd|jjksdJ|D]}|j|}||j |qf| S)a  Put a ragged batch onto the inference engine. This will perform one forward and return a Tensor of the shape [len(batch_uids), *output_shape]. Logits for the non-final tokens are not calculated. Arguments: batch_uids: Iterable of uids for the batch on the host batch_tokens: Iterable of token tensors for the batch on the host do_checks: Check schedulability when it is set to True. You can skip this check for better performance when it has already been completed. cSsg|]}t|qSr)len).0tokensrrr {sz)InferenceEngineV2.put..)r<r) can_scheduler Successr r(clearziprget_or_create_sequencermaybe_allocate_kvnumel pre_forwardinsert_sequencefinalize prepare_batchforwardshapecurrent_sequences get_sequence post_forward maybe_free_kv) rr:r;r< token_lensschedule_checkuidr? host_seq_desclogitsrrrputks(      zInferenceEngineV2.putrTmax_request_tokenscCsL|j|}|dur|jj|jjjkrdSt}|j|||\}}||fS)a Determine the number of tokens and KV blocks to reserve for a given request. Given a UID (this UID may not be recognized by the model yet), this will return the number of tokens and blocks to reserve for the request. Arguments: uid (int): The UID of the sequence (as tracked by the scheduling entity). If this is a new sequence (with a UID unknown to the inference engine), then an empty placeholder is created to pass to the occupancy logic. n_tokens (int): The number of tokens to hypothetically send. Returns: Tuple[int, Optional[int]]: Tuple of free kv blocks and the number of blocks required to schedule the sequence. N)rr) rrOn_tracked_sequencesrr'max_tracked_sequencesr rget_kv_requirements)rrTrXmax_request_blocksseq_desc req_tokens req_blocksrrrquerys zInferenceEngineV2.queryuidslengthsc Cs|jj}|jj}d}d}t||jjjkrtjSt ||D]0\}}|j |} | dur2|d7}t } |j | ||\} } | |krEtjS||7}|| 8}q||jjjkrXtjS||jjjkrbtjStjS)a| Dry run a batch to determine if it can be scheduled. Placeholder sequences will be created for any UIDs that are unknown to the inference engine. Arguments: uids (Iterable[int]): Iterable of UIDs for the batch lengths (Iterable[int]): Iterable of lengths for each sequence of the batch. This lengths corresponds to the number of tokens to send in the hypothetical forward; history tokens will be determined via UID lookup and future tokens are disregarded. Returns: bool: True if the batch can be scheduled, False otherwise. rNr)rrYrr=rr'max_ragged_sequence_countr BatchSequenceLimitExceededrDrOr rr[KVCacheLimitExceededrZEngineSequenceLimitExceededmax_ragged_batch_sizeBatchTokenLimitExceededrB) rrarbcur_seqsrr_ batch_lenrTlengthr] sched_len sched_blocksrrrrAs*   zInferenceEngineV2.can_schedulecCs$|j|}|dur dS|j|S)zQ Get the remaining capacity of the last block already allocated. Nr)rrOrget_remaining_block_capacity)rrTr]rrrrns  z.InferenceEngineV2.get_remaining_block_capacitycCs|j|dS)z Remove all state associated with a sequence from the inference engine. Arguments: uid (int): The UID of the sequence to flush. N)rflush_sequence)rrTrrrflushszInferenceEngineV2.flush save_pathcCst||jj|jj}t||jj|jj}t|jj|t |jj t |d|jjdkrAt |jj t tj|dddSdS)z~ Serialize the model to a file. Arguments: path (str): Path to the file to serialize to. wrzds_model_config.pklwbN)rrtp_rankr3rtorchsaveflattened_paramsjsondumpflattened_param_metadataopenpicklerr/pathjoin)rrqparam_file_namemetadata_file_namerrr serializes $zInferenceEngineV2.serialize)T)__name__ __module__ __qualname__r__annotations__rr propertyruTensorrr.rrrr+r#rboolrWrr`r rArnrpstrrrrrrrs:  31  r)"r/rxr|typingrrrudeepspeed.commcommr7deepspeed.acceleratorrdeepspeed.comm.commrmodel_implementationsrloggingrraggedr r r scheduling_utilsr r (model_implementations.flat_model_helpersrr*model_implementations.inference_model_baser config_v2rINFERENCE_MODEL_TIMERrrrrrs"