o پi@sNddlmZ ddlZddlZddlZddlmZmZmZddl m Z m Z m Z m Z ddlZddlmZmZe rCddlmZddlmZddlmZmZdd lmZmZmZmZdd lmZdd l m!Z!e"e#Z$e!Z%Gd d d Z&GdddZ'GdddZ(Gddde Z)GdddZ*GdddZ+Gddde+Z,GdddZ-dS)) annotationsN)EmptyFullQueue) TYPE_CHECKINGList NamedTupleOptional)HiCacheStorageConfigHiCacheStorageExtraInfo)BaseTokenToKVPoolAllocator) HostKVCache)get_tensor_model_parallel_rank$get_tensor_model_parallel_world_size)get_attention_dp_rankget_attention_tp_rankget_attention_tp_sizeis_dp_attention_enabled)MLATokenToKVPool)get_device_modulec@s6eZdZd ddZdddZddd Zed d Zd S)LayerLoadingEvent num_layersintcCs(||_ddt|D|_t|_dS)NcSsg|]}tqS) device_moduleEvent.0_rrX/home/ubuntu/.local/lib/python3.10/site-packages/sglang/srt/managers/cache_controller.py 6z.LayerLoadingEvent.__init__..) _num_layersrange load_eventsrr start_eventselfrrrr__init__4szLayerLoadingEvent.__init__ layer_indexcCs0d|kr |jksJJ|j|dSNr)r"r$recordr'r)rrrcomplete9szLayerLoadingEvent.completecCst|j|dSN)rcurrent_stream wait_eventr$r,rrrwait=szLayerLoadingEvent.waitcCs |jdSN)r$r'rrr finish_event@s zLayerLoadingEvent.finish_eventNrr)r)r)__name__ __module__ __qualname__r(r-r1propertyr5rrrrr3s    rc@s:eZdZdddZddZddd Zdd d Zd dZdS)LayerDoneCounterrrcs6|_d|_fddt|jD|_d|_d|_dS)Ncsg|]}tqSr)rrrrrr Jr!z-LayerDoneCounter.__init__..r3)r num_countersr#eventsproducer_indexconsumer_indexr&rr=rr(Fs  zLayerDoneCounter.__init__cCs2|jd|j|_|j|jjsJd|jS)Nz:Producer finish event should be ready before being reused.)r@r>r?r5queryr4rrrupdate_producerNsz LayerDoneCounter.update_producerindexcCs ||_dSr.)rA)r'rErrr set_consumerW zLayerDoneCounter.set_consumer thresholdcCs$|jdkrdS|j|j|dSr*)rAr?r1)r'rHrrr wait_untilZs zLayerDoneCounter.wait_untilcCsd|_d|_dSr2)r@rAr4rrrreset_s zLayerDoneCounter.resetNr6)rEr)rHr)r7r8r9r(rDrFrIrJrrrrr;Es    r;c@s6eZdZdZ ddd d ZedddZdddZdS)CacheOperationrN host_indices torch.Tensordevice_indicesnode_idrpriority Optional[int]cCsN||_||_|g|_d|_tj|_tjd7_|dur!||_dS|j|_dS)NrB)rLrNnode_idsdatarKcounteridrP)r'rLrNrOrPrrrr(hszCacheOperation.__init__opsList[CacheOperation]returncCst|dksJt|dkr|dStdd|D}tdd|D}g}tdd|D}|D]}||jq3t||d|}||_|S) NrrBcSg|]}|jqSr)rLroprrrr z,CacheOperation.merge_ops..cSrYr)rNrZrrrr r\css|]}|jVqdSr.rPrZrrr sz+CacheOperation.merge_ops..r3)lentorchcatminextendrRrK)rVrLrNrRrPr[ merged_oprrr merge_opsys zCacheOperation.merge_opsothercC |j|jkSr.r]r'rfrrr__lt__ zCacheOperation.__lt__r.)rLrMrNrMrOrrPrQ)rVrWrXrK)rfrK)r7r8r9rTr( staticmethodrerirrrrrKds  rKc@s&eZdZUded<ded<ded<dS) HiCacheAckzdevice_module.Eventr%r5 List[int]rRN)r7r8r9__annotations__rrrrrls  rlc@sReZdZdZ ddd d Zdd d ZdddZddddZddddZddZ dS) TransferBufferzW Overlapping buffer preparation and transfer operations to improve throughput. r< buffer_countrmax_buffer_sizerXNonecCs||_t|d|_||_dS)N)maxsize) stop_eventrbuffersrr)r'rurqrrrrrr(s  zTransferBuffer.__init__boolcC |jSr.)rvfullr4rrrryrGzTransferBuffer.fullcCrxr.)rvemptyr4rrrrzrGzTransferBuffer.emptyTrBc Cs~|js=z |jj|||dWdSty|sYdSYqty5}z t|WYd}~nd}~ww|jrdSdSNblocktimeout)ruis_setrvputr Exceptionloggererror)r'itemr}r~errrrs  zTransferBuffer.putOptional[CacheOperation]c CsVz |jj||dWStyYdSty*}z t|WYd}~dSd}~wwr{)rvgetrrrr)r'r}r~rrrrrs zTransferBuffer.getcCs|jjdSr.)rvqueueclearr4rrrrszTransferBuffer.clearN)r<rp)rqrrrrrXrsrXrw)TrBrXrs)rXr) r7r8r9__doc__r(ryrzrrrrrrrros    roc@s,eZdZdZ   ddd d ZdddZdS)StorageOperationrNrLrM token_idsrm last_hash Optional[str] hash_valueOptional[List[str]] prefix_keyscCsJ||_||_||_d|_|dur|ng|_||_tj|_tjd7_dS)NrrB) rLrrcompleted_tokensrrrrTrU)r'rLrrrrrrrr(szStorageOperation.__init__rf'StorageOperation'cCrgr.)rUrhrrrrirjzStorageOperation.__lt__)NNN) rLrMrrmrrrrrr)rfr)r7r8r9rTr(rirrrrrs rcs@eZdZ  ddfd d ZdddZddZdddZZS)PrefetchOperationN request_idstrrLrMrrmrrrrcs8||_t|_d|_t|_tj ||||ddS)NFr) r threadingLock_lock_terminated_flagtime monotonic start_timesuperr()r'rrLrrr __class__rrr(s   zPrefetchOperation.__init__ num_tokensrcCsT|j|jr WddS|j|7_ WddS1s#wYdS)NFT)rrr)r'rrrr increments$zPrefetchOperation.incrementcCs2|j d|_WddS1swYdS)NT)rrr4rrrmark_terminates"z PrefetchOperation.mark_terminaterXrwcCs|jSr.)rr4rrr is_terminatedszPrefetchOperation.is_terminatedNN) rrrLrMrrmrrrr)rrr)r7r8r9r(rrr __classcell__rrrrrs rc@sZeZdZ        dodpddZddZd d!Z   dqdrd"d#Zd$d%Z  dsdtd&d'Zd(d)Z  *dudvd2d3Z dwd5d6Z  *dudxd8d9Z dydd?Zd{d@dAZd|d}dEdFZ  dsd~dNdOZdPdQZddRdSZ ddTdUZddVdWZdXdYZdZd[Zdd\d]Zdd_d`ZdadbZ  dsddedfZdddgdhZdddidjZdkdlZdmdnZdS)HiCacheControllerwrite_through_selectiveNrrBtoken_to_kv_pool_allocatorr mem_pool_hostr page_sizertp_grouptorch.distributed.ProcessGroupload_cache_eventthreading.Event write_policyr io_backendstorage_backendrprefetch_threshold model_namestorage_backend_extra_configOptional[dict]pp_rankpp_sizec CsL||_||_||_||_||_||_||_d|_d|_ d|_ | |_ | |_ |j |_|j|_t|_|jj|_|jj|_t|j|_|j|j|dvrUtd|g|_g|_g|_g|_t|_t|j|_ t|jddd|_!t"#|_$t"#|_%|durz |j&|| | | dWdSty}ztd||d}~wwdS) NF) write_throughr write_backzInvalid write policy: d)rqrr)rrrrz"Failed to create storage backend: )'rmem_pool_device_allocator get_kvcachemem_pool_devicerrrrenable_storagerstorage_backend_typerr_generic_page_get page_get_func_generic_page_set page_set_funcrrstorage_stop_eventdevice layer_numr;layer_done_counterregister_layer_transfer_counter ValueError load_queue write_queueack_load_queueack_write_queueruro write_buffer load_bufferrStream write_stream load_streamattach_storage_backend)r'rrrrrrrrrrrrrrrrrr(sZ          zHiCacheController.__init__cCs||jsJ|jr Jtj|jdd|_tj|jdd|_t |_ t |_ t |_ t |_ t |_|j|jdS)zStart storage prefetch/backup threads and their queues. This is used by runtime attach, and also by reset when storage is enabled. TtargetdaemonN)rrrrThreadprefetch_thread_funcprefetch_threadbackup_thread_func backup_threadrprefetch_queue backup_queueprefetch_revoke_queueack_backup_queuehost_mem_release_queuestartr4rrr_start_storage_threadsIs  z(HiCacheController._start_storage_threadsc Cs |jz#t|dr|jdt|dr|jdt|dr'|jdWn ty1Ynwg}t|dr?||j t|drJ||j t|drU||j |D]}z|j dd WqWtyjYqWwd d |D}|rt d d d |DtddS)z}Stop storage prefetch/backup threads and drain internal queues. Caller should ensure no in-flight requests. rNrprefetch_bufferrrprefetch_io_aux_threadr)r~cSs"g|] }t|dddr|qS)is_alivecSsdS)NFrrrrrszDHiCacheController._stop_storage_threads...)getattrrtrrrr s"z;HiCacheController._stop_storage_threads..z2Failed to stop HiCache storage threads cleanly: %scSsg|] }t|dt|qS)name)rreprrrrrr sz/Failed to stop HiCache storage threads cleanly.)rsethasattrr put_nowaitrrrappendrrrjoinrr RuntimeError)r'threadsraliverrr_stop_storage_threadsasB                z'HiCacheController._stop_storage_threadsc Cs||jrtdz|Wnty}ztd|d}~ww||_ddlm}||_||||_|jj o;|jj dk|_ ddl m }z|||j|j|_|j|jd|_t||j|_tdtd|jj|jj|_d |_d|_tjj|jd |_|jd krdd l m!}tj"|j} || d d|_#|j$|_%|j&|_'|jdvs|jdkrt(|jj)*ddr|j+|_%|j,|_'|j-.|/WdSty=z|Wn tyYnwzt0|drz tj1|j#Wn tyYnwd|_#Wn tyYnwzt0|dr|jdurt0|jdr|j2Wn ty*Ynwd|_d|_d|_|j$|_%|j&|_'w)zAttach (enable) storage backend at runtime. Requirement: no in-flight requests. This call is expected to run on the scheduler thread (control path), not concurrently with prefetch/backup. z!Storage backend already attached.zTCannot attach storage backend: previous detach did not stop storage threads cleanly.Nr) get_hash_str)StorageBackendFactoryTg?)grouprB)create_custom_parallel_groupgloo) group_ranksbackend)hf3fsmooncakeeicnixldynamic interface_v1prefetch_tp_grouprcloseF)3rrrrr$sglang.srt.mem_cache.hicache_storager_generate_storage_configstorage_config is_mla_modeltp_rank backup_skipsglang.srt.mem_cache.storagercreate_backendrrregister_mem_pool_hostmaxrrrsizerprefetch_capacity_limitstorage_batch_sizeprefetch_tokens_occupiedr` distributedget_world_sizer tp_world_size%sglang.srt.distributed.parallel_staterget_process_group_ranksr rrrrrw extra_configr_page_get_zero_copy_page_set_zero_copyrrrrdestroy_process_groupr ) r'rrrrrrrrrrrrrs                 z(HiCacheController.attach_storage_backendc Csz|Wnty}z td|td|d}~wwz"t|dr?|jdur?z tj |jWn ty;Ynwd|_Wn tyIYnwzt|dr`|j dur`t|j dr`|j WntyotdYnwd|_ d|_ d|_ |j|_|j|_|jdS) zDetach (disable) storage backend at runtime. Requirement: no in-flight requests. This will stop storage threads and release the backend instance (best-effort close). zStop storage threads failed: %sz,Stop storage threads failed; detach aborted.Nr rr z(Failed to close storage backend cleanly.F)rrr exceptionrrr r`rr"rr rrrrrrrr)r'rrrrdetach_storage_backendsL           z(HiCacheController.detach_storage_backendc Csjtrt|_t|_t|_n t|_t|_d|_t |j t }t |j|j|j |j||jjdk||dS)Nr page_first)rtp_sizerrris_page_first_layoutrr)rrrrr&rdp_rankrr isinstancerrr rrrlayout)r'rris_mla_backendrrrr 9s$   z*HiCacheController._generate_storage_configcCs|j|j|j|j|j|j|j|j |j rM|j |j |jj|jj|jj|jj|j|j|j rxtj|jdd|_ tj|jdd|_ |j |j dSdS)NTr)rurrrrrrrrrrrrrrrrrrrrrrrr4rrrrJVs6                 zHiCacheController.resetr3rNrMrPrQrOrXOptional[torch.Tensor]cCs>|jt|}|durdS|jt||||||S)zF Back up KV caches from device memory to host memory. N)rallocr_rrrK start_writing)r'rNrPrOrLrrrwriteus  zHiCacheController.writerscCst|jdkr dSt|j}||\}}|jt}t}|t |j /| |j |j |j|||j||jrL||j |jrU||j Wdn1s_wY|jt|||jdSr*)r_rrKre move_indicesrrrr+streamrr1rbackup_from_device_all_layerrris_cuda record_streamrrrlrR)r'r[rLrNr%r5rrrr.s*      zHiCacheController.start_writingrLcCs6|jt|}|durdS|jt|||||S)zC Load KV caches from host memory to device memory. N)rr-r_rrrK)r'rLrPrOrNrrrloads  zHiCacheController.loadr[rKcCs|j|j}}|jdkr|js|j|jdd}||fS|jdkrF|jjdkr8|}| \}}|| d|fS|jjdkrD||fSdS|jdkrQ||fSt d ) NkernelT) non_blockingdirect layer_firstrpage_first_direct kernel_ascendzUnsupported io backend) rLrNrr3torrr*cpusort index_selectr)r'r[rLrNidxrrrr0s         zHiCacheController.move_indicesc Cst|jdkr dS|j}t|j}||\}}|j|jj|}|j t |j :|j |j t|jD]}|j|j||||j||q>|jr[||j |jrd||j Wdn1snwY|jt|j |j|jd|S)Nrr3)r%r5rR)r_rrrDrKrer0rr?r%r+rr1rr1r#rrload_to_device_per_layerrrr-r3r4rrrlr5rR)r' producer_idr[rLrNproducer_eventirrr start_loadingsB        zHiCacheController.start_loadingcCs|j|t|Sr.)rfreer_)r'rNrrr evict_devices zHiCacheController.evict_deviceT backup_onlyrwcCs |std|j|t|S)Nz.Other eviction policies are not supported yet.)rrrFr_)r'rLrHrrr evict_hosts zHiCacheController.evict_hostrnew_input_tokensrmrrrrcCs t|||||}|j||S)zI Prefetch KV caches from storage backend to host memory. )rrr)r'rrLrJrr operationrrrprefetchs   zHiCacheController.prefetchcCs||j|jfSr.)rrrr'rKrrrterminate_prefetch s z$HiCacheController.terminate_prefetchcCs8|dkrdS||jj}|D]}|j|qdSr*)numelsplitrrrr)r'rLpagespagerrrappend_host_mem_releases z)HiCacheController.append_host_mem_releasecCsh|j|||}d}tt|D]}||s'td|jd||dn||j7}q||dS)NrPrefetch operation  failed to retrieve page .) r batch_get_v1r#r_rwarningrrr)r'rK hash_valuesrL extra_inforesultsincrDrrrr s z%HiCacheController._page_get_zero_copycsfdd|D}j||}|durdStt|D]1}||dur6td|jd||ddSj||j ||| j sMdSqdS)Ncsg|]}jqSr)rget_dummy_flat_data_pagerr4rrr 's z7HiCacheController._generic_page_get..rTrUrV) r batch_getr#r_rrXrrset_from_flat_data_pagerr)r'rKrYrLrZdummy_page_dst page_datarDrr4rr&s(    z#HiCacheController._generic_page_getcCs|j}tdt|j|jD]L}|j|||j}|j||j|t||j}|j}t|d}| |||||j|t||jkrM| dS|rYt|dkrY||7}q dSNrr) rr#r_rrrLrrr rr)r'rKrrD batch_hashesbatch_host_indicesprev_completed_tokensrZrrr_page_transfer<s$ z HiCacheController._page_transfercCsp|js6z |jjddd}|durWq||||j|jdWn ty.Yqw|jrdSdS)zN Auxiliary function conducting IO operations for prefetching. TrBr|N) rrrrrfrSrLrrrMrrrprefetch_io_aux_funcSs   z&HiCacheController.prefetch_io_aux_funccCs|j|jkrdSdS)zb Rate limit the prefetching operations to avoid overwhelming the storage backend. TF)rrr4rrrprefetch_rate_limitedds z'HiCacheController.prefetch_rate_limitedtuple[list[str], int]cCs|j}|j}|jr|jnd}d}g}tdt||j|jD]h}t||j|jt|}|||} g} tdt| |jD]} | | | | |j|}| |q@t |d} |j | | } || d| || |j7}| t| kr|||fS|rt|dkr|| 7}q ||fSrb)rrrcopyr#r_rrrbrrr r batch_existsrc)r'rKrtokens_to_fetchrstorage_query_countrrend batch_tokensrcrDrZ hit_page_numrrr_storage_hit_queryns:    z$HiCacheController._storage_hit_querycCsvt|_tj|jdd|_|j|jr|j sz|j j ddd}|dur+Wq| |\}}|j dkrPtj|tjd}tjj|tjjj|jd|}||jkrp|j|j||jtd|jd |d n2|d||j|_ ||j|d|jd||_td t!|j d |jd |j|Wn t"yYqw|jr|j rdSdS)zT Manage prefetching operations from storage backend to host memory. TrrBr|N)dtype)r[rzRevoking prefetch for request z due to insufficient hits (z).z Prefetching z pages for request rV)#rrrrrgrrrrrrzrrqrr`tensorrr all_reduceReduceOpMINr rrrrrrSrLrdebugrrr_r)r'rKrstorage_hit_countstorage_hit_count_tensorrrrrsV        z&HiCacheController.prefetch_thread_funcrrcCs"t||||d}|j||jS)zF Write KV caches from host memory to storage backend. )rr)rrrrU)r'rLrrrrKrrr write_storages   zHiCacheController.write_storagecs*fddtt|D}j||S)Ncs"g|] }j|jqSr)r get_data_pager)rrDrLr'rrr sz7HiCacheController._generic_page_set..)r#r_r batch_set)r'rYrLrZrSrr|rrs  z#HiCacheController._generic_page_setcCst|j|||Sr.)allr batch_set_v1)r'rYrLrZrrrr!sz%HiCacheController._page_set_zero_copycCs|j}tdt|j|jD]Q}|j|||j}|j||j|t||j}t|d}||||}|sFt dt|ddS|rRt|dkrR||7}|j |jt|7_ q dS)NrrzWrite page to storage: z pages failed.) rr#r_rrrLrr rrrXr)r'rKrrDrcrdrZsuccessrrr _page_backups" zHiCacheController._page_backupcCsl|js4z|jjddd}|durWq|js|||j|Wn ty,Yqw|jrdSdS)zO Manage backup operations from host memory to storage backend. TrBr|N) rrrrrrrrrrMrrrrs   z$HiCacheController.backup_thread_func)rrNrNNrrB)rr rr rrrrrrrrrrrrrrrrrrrrrr)rNN)rrrrrrrrr)rrrrr2)rNrMrPrQrOrrXr,r)rLrMrPrQrOrrXr,)r[rK)rXr)rNrMrXr)T)rLrMrHrwrXr) rrrLrMrJrmrrrrrXr)rLrMr.r)rXri) rLrMrrmrrrrrXr) r7r8r9r(rrrr$r rJr/r.r5r0rErGrIrLrNrSr rrfrgrhrqrrzrr!rrrrrrrsh  P1 u7 "    '        6   r). __future__rloggingrrrrrrtypingrrrr r`r r r sglang.srt.mem_cache.allocatorr %sglang.srt.mem_cache.memory_pool_hostr sglang.srt.distributedrrsglang.srt.layers.dp_attentionrrrr sglang.srt.mem_cache.memory_poolrsglang.srt.utilsr getLoggerr7rrrr;rKrlrorrrrrrrs4       )+