^jqQdZddlZddlmZddlmZddlZddlmZddl m Z dd l m Z dd l mZmZmZmZdd lmZd eed eeeeeffdZGddZy)uCentralized offloading logic for continuous batching. Handles two offloading strategies when the GPU KV cache is full: 1. CPU offloading: copy the KV cache to a pre-allocated pinned CPU buffer, preserving exact request state. 2. Soft reset: discard the KV cache and re-prefill from scratch when the request is re-scheduled. This incurs no data transfer overhead, but we need to re-run prefill over all initial + generated tokens (so more compute overhead). The CPU swap pool is a static set of pinned tensors allocated once at init (like vLLM/SGLang). Blocks are tracked with a simple free set — no dynamic allocation or deallocation of tensors ever happens at runtime. N) nullcontext)chain)is_psutil_available)PagedAttentionCache)DistributedHelper)FutureRequestState RequestState RequestStatuslogger) Schedulerindicesreturncg}|s|S|dx}}d}tdt|D]0}|||dzk7r|j||||z f|||}}||}2|j||t||z f|S)zGroups an index list into (start_index, offset, length) runs of consecutive values, so scattered block copies can be performed as a few slice copies.rr)rangelenappend)rrunsstartprevoffsetis /var/www/ramen.bs-engineer-server.com/venv/lib/python3.12/site-packages/transformers/generation/continuous_batching/offloading_manager.pycontiguous_runsr&s D  1:ED F 1c'l # 1: ! KKF 3 4#AJ6Eqz   KKG v 567 KceZdZdZdedededzdedejjdzde d dfd Z dedzded e fd Z d Zd e fd Zdeed dfdZddeded dfdZddZddZdeed eefdZded eee ee ffdZy)OffloadingManageru/Manages request offloading and restoration for continuous batching. Owns a static CPU swap pool (pre-allocated pinned tensors mirroring the GPU cache layout), performs GPU↔CPU block copies, decides between CPU offloading and soft reset, and ensures cleanup on cancellation/failure/reset. cache schedulercpu_offload_space_gibNsafety_thresholdcompute_streamdistributed_helperrc||_||_||_g|_g|_g|_g|_g|_i|_i|_ |j||}tj|tjd}t|j|dj!|_|duxr|dkD}|j"dk(r|rt%j&d|ddy|j"|j(|j*|j,f} |j.D]x} |jj1tj2| |j4d |jj1tj2| |j4d zd |j(|j*|j,f} t7|j.|j8D]U\} } |j j1| j:| |j j1| j:| Wt=t?|j"|_|jd}d |jAz|jCztE|j.z}t%jFd |j"d |dz ddy)NcpudtypedeviceT)on_cpurcpu_offload_space=.1fz8 GiB is too small for even one block. No CPU offloading.)r( pin_memoryzCPU swap pool initialized: z blocks (@.2fz GiB pinned))$rr _compute_stream_cpu_key_cache_cpu_value_cache_gpu_key_views_gpu_value_views_free_cpu_blocks_request_id_to_cpu_blocks!_request_id_to_group_block_counts_compute_num_cpu_blockstorchtensorint32inttp_all_reduce_minitem_num_cpu_blocksr warning block_sizenum_key_value_headshead_dim key_cacheremptyr(zip value_cacheviewlistrnumel element_sizerinfo)selfrr r!r"r#r$num_cpu_blocksoffloading_enabledcpu_cache_shape_ block_shapek_cachev_cache cache_tensor size_in_bytess r__init__zOffloadingManager.__init__>s "-35462446+-?A&GI.556KM]^nEKKPUV"#5#G#G_c#G#d#i#i#kl2$>\CX[\C\   1 $!()>s(CD)) //1A1A5C\C\^c^l^lm kA    & &u{{?%++bf'g h  ! ! ( (_EKKdh)i j k 5++U-F-FW #EOOU5F5F G E GW    & &|w||['A B  ! ! ( ({)C D E !%U4+?+?%@ A**1- L..00<3L3L3NNQTUZUdUdQee  )$*>*>)?yZaIbcfHggs t rc |t|dznd}tr-ddl}|jj}t||z}nd}|8|6||kDrv|dz }t j d|dd|dddz dd |dd |}nE|t j d n-| |}t j d |dz d d n tddt|jjz|jjz|jjz|jjz|jjjz}|dk(r t!d||zS)z?Returns the number of blocks that can fit in the CPU swap pool.Nr0rr+r,z GiB exceeds z.0%z of total RAM (z GiB). Clamping to z GiB.u{psutil is not available — cpu_offload_space_safety_threshold cannot be enforced. Install psutil to enable the safety cap.z1Auto-sizing CPU swap pool from safety threshold: r1ztcpu_offload_space=None requires psutil to auto-size the CPU swap pool. Install psutil or pass an explicit GiB value.r/z9The number of bytes per block is 0. This is not possible.)r>rpsutilvirtual_memory availabler rB ImportErrorrrrFrCrDrEr(itemsize ValueError) rOr!r" offload_bytesr[ total_ram max_bytes clamped_gibbytes_per_blocks rr:z)OffloadingManager._compute_num_cpu_blocksysCXBc1W=>im   --/99II(889II  $)>y('73 ()>s(C=QabePfg!W-c22EkRUEVV[]!*  & NN;   "%M NNNy\cOdehNiino p&  $**&&' (jj## $jj,, -jj!!  " jj''  (  a XY Y//rc|j)tjj|jSt S)zdReturns a context manager that runs enclosed ops on the compute stream, or a no-op when none is set.)r2r;cudastreamrrOs r _stream_ctxzOffloadingManager._stream_ctxs1:>:N:N:Zuzz  !5!56m`k`mmrc |j}|j}|sy|jj}t d|D}t |j }g}||kDr}|r{|t |z dkDrj|j\}}|j|||z}||jj|jz }||kDr|r|t |z dkDrj|sy|j|j|} |D]}|j} | | vrd|_ |jtj k(r|j"t |j$t |j&zk(r;|xj"dzc_|j&xs |j$d} | g|_n|j*dd|_tj,|_|} n%|j/} tj0|_|j3| |j5|  d|_t9j:t<j>ro|jj}t9j@dt |dt | d t |t | z d t |d |d t |S) aEvict enough active requests that, at the next batch, every remaining starved request can allocate the blocks it needs. Victims are taken from the starved requests reported by the scheduler, newest first, so the batch that was just scheduled is never touched and the demand directly bounds the amount of offloading. Tries CPU offloading first; victims that do not fit in the pool are soft reset. Returns the number of evicted requests.rc3&K|] \}}| yw)N).0rS blocks_neededs r z5OffloadingManager.offload_requests..sC'7q-]Csrr.NTz Offloaded z requests (z to CPU, z soft reset): z starved requests remain for z free blocks.)!r starved_requestsrget_num_free_blockssumractive_requestspopr blocks_in_use request_idreverse_offload_to_cpuallocated_blocks_statusr DECODINGposition_offsetinitial_tokensgenerated_tokensremaining_prefill_tokenstokens_to_processPENDING!create_equivalent_initial_requestFINISHEDfinish_requestadd_waiting_requestblock_new_requestsr isEnabledForloggingINFOrN) rOr starved free_blocksdemand num_activevictimsstatero cpu_offloadedrwlast_true_token new_states roffload_requestsz"OffloadingManager.offload_requestss NN ,,jj446 C7CC223 &({"w:G 3Lq3P#*;;= E= NN5 ! m #F 4::33E4D4DE EK {"w:G 3Lq3P ,,W5  5E))J]*)*&==M$:$::,,E4H4H0ICPUPfPfLg0gg--2-+0+A+A+YUEYEY[]*^:I9J6:?9P9PQR9S6!. 5 5 ! !CCE - 6 6  $ $Z 0  ) )) 41 54(, $   w|| ,**88:K KKS\N+c-6H5ISVW^S_befsbtStRuv #G ~-J;-Wdf 7|rrequests_in_batchc <|j}g}g}|D]}}|j}|js|jj |j }|j j |j }|j|d} t|D]a\} } |j| jj|j g} |j| d| t| t| } cd|_| |_|jr,|xj |j"|j$zz c_t'j(d|j dt|j*dt|j,d|syt|} t/|} |j}|j15t3j4|t2j6j9|j:d }| |j$|j<|j>f}t3j@||jB|j: }tE|jF|jH}tE|jJ|jL}tO||D]B\}}| D]%\}}}||||zjQ||||zd '|jSd||D dddtU|jV|z|_+y#1swY'xYw) zRestore KV caches from CPU for any CPU-offloaded requests in the scheduled batch. Indices are accumulated per group across all requests, then copied in one batched operation per layer.rNFzRestored CPU-offloaded request z with z prefill tokens and z generated tokens.r(T non_blockingr'),rris_cpu_offloadedr8rurwr9extend enumerategroup_cache_managers block_tablegetmaxrrzallow_block_sharingcomplete_blocksr}rCr debugr~rrrjr; as_tensorlongtor)rDrErGr(rr3r4r5r6rHcopy_ index_copy_sortedr7)rOrrall_cpu_indicesall_gpu_indices future_stater cpu_indices group_countsmax_allocated_blocks group_idxn gpu_blocksrgpu_ids staging_shapestaging cpu_caches gpu_views cpu_cachegpu_viewrrlengths rrestore_scheduled_requestsz,OffloadingManager.restore_scheduled_requestss %'%'- L &&E))88<  /      :oooUZZHKKELLgkKlG 0 0%2K2KU^^\Mkk-u{{5<<XGt22D4I4IJJd1143H3HII'*:y'A :# 8-1r)E66FVf_5;;IeeV\n<]lp;qr$$Q9 : :!'t'<'<'N O : :s D"LLr keep_unsortedc|jr@|j|jd|_|s|jj yyy)z=Free CPU blocks for a single request (e.g., on cancellation).FN)r_return_cpu_blocksrwr7sort)rOrrs rfree_request_cpu_cachez(OffloadingManager.free_request_cpu_cache,sF  ! !  # #E$4$4 5%*E " %%**,! "rc|jjjD]}|j|d|jj y)zPFree all CPU-offloaded caches in the waiting queue (e.g., on fail_all or reset).T)rN)r waiting_requestsvaluesrr7r)rOrs rfree_all_waiting_cpu_cachesz-OffloadingManager.free_all_waiting_cpu_caches4sK^^44;;= CE  ' 'T ' B C ""$rc|j|jj|jjt t |j |_y)z8Reset CPU offloading state for a new generation session.N)rr8clearr9rKrrAr7ris rresetzOffloadingManager.reset:sJ ((* &&,,. ..446 $U4+?+?%@ Arrc Dg}g}t|j}|D]}g}g}|jjD]S}|jj |j g} |j| |jt| U|svt|t|z|ks|j|||f|j||s tSt|} |jd| } |j| d|_t| } |j5tj|tjj|jj d} t#|j$|j&}t#|j(|j*}t-||D]A\}}|j/d| }| D]%\}}}||||zj1||||zd'C dddd}|D]\\}}}| ||t|z|j2|j <||j4|j <d|_|t|z }^|Dchc]\}}}|j c}}S#1swYxYwcc}}w)aCopy the KV cache blocks of as many victims as fit in the CPU swap pool from GPU to the pool, in one batched, non-blocking copy per layer. Returns the request ids that were offloaded. All transfers are enqueued on the compute stream with pinned destinations, so the host never waits on them: correctness is guaranteed by stream ordering, since restores and cache writes go through the same stream. NrTrr)rr7rrrrrwrrsetrrjr;rrrr)rr5r6r3r4rH index_selectrr8r9r)rOr offloadedrfree_pool_blocksr gpu_indicesgroup_block_countscmblocksrrrrrrrrgathered_blocksrrrrSs rryz!OffloadingManager._offload_to_cpuAsFH %'t445 4EK!# jj55 7++E,<,>[UAq  >>%  $?s)CJ5JJrwc|jj|}|jj|}|jj |||fS)zr:rjrrKr rr boolrrrrstrrytuplerrmrrrr7s 9 "9 9  %t| 9  9  ))D0 9 .9  9 v30UT\30]b30gj30jn>#>@8PDASrrrrmrrrs] " (&*MM T#Y4c3m0D+E"E%E%r