File size: 34,648 Bytes
3844f4b |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624 625 626 627 628 629 630 631 632 633 634 635 636 637 638 639 640 641 642 643 644 645 646 647 648 649 650 651 652 653 654 655 656 657 658 659 660 661 662 663 664 665 666 667 668 669 670 671 672 673 674 675 676 677 678 679 680 681 682 683 684 685 686 687 688 689 690 691 692 693 694 695 696 697 698 699 700 701 702 703 704 705 706 707 708 709 710 711 712 713 714 715 716 717 718 719 720 721 722 723 724 725 726 727 728 729 730 731 732 733 734 735 736 737 738 739 740 741 742 743 744 745 746 747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762 763 764 765 766 767 768 769 770 771 772 773 774 775 776 777 778 779 780 781 782 783 784 785 786 787 788 789 790 791 792 793 794 795 796 797 798 799 800 801 802 803 804 805 806 807 808 809 810 811 812 813 814 815 816 817 818 819 820 821 822 823 824 825 826 827 828 829 830 831 832 833 834 835 836 837 838 839 840 841 842 843 844 845 846 847 |
# SPDX-FileCopyrightText: Copyright (c) 2025, NVIDIA CORPORATION. All rights reserved.
# SPDX-License-Identifier: Apache-2.0
import functools
import gc
import math
from typing import Any
import torch
from transformers import CacheLayerMixin
from transformers.cache_utils import Cache
def ceil_int_div(a: int, b: int) -> int:
return (a + b - 1) // b
def float_ceil(a: float):
return math.ceil(a)
def _aux_potential_eviction(
vals_for_replacement: torch.Tensor,
to_be_evicted_table_block_id: torch.Tensor,
to_be_evicted_position_within_block: torch.Tensor,
to_be_evicted_mask: torch.Tensor,
block_table: torch.Tensor,
blocks: torch.Tensor,
page_batch_index: torch.Tensor,
last_table_block_id: torch.Tensor,
next_position_within_block: torch.Tensor,
):
"""Adding a new element to KV cache may lead to eviction of the last element in the DMS sliding window."""
# For each batch element the block table contains a list of blocks allocated for this batch element
block_ids = block_table[page_batch_index, to_be_evicted_table_block_id]
# Override the last element of the sliding window with the new element if the last element of the sliding window
# is marked for the eviction and the window is full
blocks[block_ids, to_be_evicted_position_within_block, :, :] = (
blocks[block_ids, to_be_evicted_position_within_block, :, :] * (1 - to_be_evicted_mask[:, None, None])
+ vals_for_replacement[:, 0, None, :] * to_be_evicted_mask[:, None, None]
)
# Otherwise write the new element to the next position within the last allocated block
block_ids = block_table[page_batch_index, last_table_block_id]
blocks[block_ids, next_position_within_block, :, :] = blocks[
block_ids, next_position_within_block, :, :
] * to_be_evicted_mask[:, None, None] + vals_for_replacement[:, 0, None, :] * (
1 - to_be_evicted_mask[:, None, None]
)
@torch.compile()
def _aux_update_single(
key_states: torch.Tensor,
value_states: torch.Tensor,
eviction_info: torch.Tensor,
recent_info: torch.Tensor,
recent_info_position: torch.Tensor,
block_table: torch.Tensor,
key_blocks: torch.Tensor,
value_blocks: torch.Tensor,
cache_seq_lenghts: torch.Tensor,
page_batch_index: torch.Tensor,
) -> torch.Tensor:
# page_batch, seq_len, head_dim = key_states.size()
# page_batch, seq_len = eviction_info.size()
# page_batch_index is a tensor of shape (page_batch,): 0, 1, 2, ... page_batch - 1
block_size = key_blocks.size(1)
# `recent_info_position` points to the next position in the sliding window; when the sliding window is full,
# it points to the first position. Not-filled elements are not zeroed out and not marked for eviction
# (see recent_info initialization).
eviction_candidate_info_position = recent_info_position % recent_info.size(1)
eviction_candidate_info = recent_info[
page_batch_index, eviction_candidate_info_position
] # Note that this is zeroed out in the beginning
# `eviction_candidate_info[:, 1]` is 1 when the element is marked for eviction and 0 otherwise
# `block_table[eviction_candidate_info[:, 0] // block_size]` is the block id where the element resides
# and `eviction_candidate_info[:, 0] % block_size` is the position (offset) within the block
to_be_evicted = eviction_candidate_info[:, 1] == 1
to_be_evicted_kv = to_be_evicted.to(key_blocks.dtype)
to_be_evicted_int = to_be_evicted.to(torch.int32)
to_be_evicted_position = eviction_candidate_info[:, 0]
to_be_evicted_table_block_id = to_be_evicted_position // block_size
to_be_evicted_position_within_block = to_be_evicted_position % block_size
last_table_block_id = cache_seq_lenghts // block_size
next_position_within_block = cache_seq_lenghts % block_size
_aux_potential_eviction(
vals_for_replacement=key_states,
to_be_evicted_table_block_id=to_be_evicted_table_block_id,
to_be_evicted_position_within_block=to_be_evicted_position_within_block,
to_be_evicted_mask=to_be_evicted_kv,
block_table=block_table,
blocks=key_blocks,
page_batch_index=page_batch_index,
last_table_block_id=last_table_block_id,
next_position_within_block=next_position_within_block,
)
_aux_potential_eviction(
vals_for_replacement=value_states,
to_be_evicted_table_block_id=to_be_evicted_table_block_id,
to_be_evicted_position_within_block=to_be_evicted_position_within_block,
to_be_evicted_mask=to_be_evicted_kv,
block_table=block_table,
blocks=value_blocks,
page_batch_index=page_batch_index,
last_table_block_id=last_table_block_id,
next_position_within_block=next_position_within_block,
)
final_position = to_be_evicted_position * to_be_evicted_int + (1 - to_be_evicted_int) * (cache_seq_lenghts)
previous_recent_info_position = (recent_info_position + recent_info.size(1) - 1) % recent_info.size(1)
# Update the eviction info for the previous element in the sliding window (if present)
recent_info[page_batch_index, previous_recent_info_position, 1] = (
eviction_info[:, 0] * (cache_seq_lenghts > 0).to(torch.int32)
).to(torch.int32)
# No info about eviction yet for the new element
recent_info[page_batch_index, eviction_candidate_info_position, 1] = 0
recent_info[page_batch_index, eviction_candidate_info_position, 0] = final_position
recent_info_position[...] += 1
cache_seq_lenghts[...] = cache_seq_lenghts + (1 - to_be_evicted_int)
# At the beginning of this function call block_table[cache_seq_lenghts // block_size] points to a block with
# at least one free position; need to maintain this invariant by detecting filled blocks
requires_free_page = torch.logical_and((cache_seq_lenghts % block_size) == 0, to_be_evicted_int == 0)
return requires_free_page
def _aux_get_recent_position_size(cache_seq_lenghts: torch.Tensor, dms_window_size: int) -> torch.Tensor:
return torch.clamp(cache_seq_lenghts, max=dms_window_size)
def _aux_get_first_recent_position(
recent_info_position: torch.Tensor,
cache_seq_lenghts: torch.Tensor,
dms_window_size: int,
) -> torch.Tensor:
return recent_info_position - _aux_get_recent_position_size(
cache_seq_lenghts=cache_seq_lenghts, dms_window_size=dms_window_size
)
def _aux_write_kv(
block_table: torch.Tensor,
blocks: torch.Tensor,
write_positions: torch.Tensor,
values: torch.Tensor,
page_batch_index: torch.Tensor,
):
page_batch, chunk_len = write_positions.size()
block_size = blocks.size(1)
block_table_id = write_positions // block_size
position_within_block = write_positions % block_size
block_id = block_table[page_batch_index[:, None], block_table_id]
assert (block_id != -1).all()
blocks[block_id, position_within_block, :, :] = values[:, :, None, :]
@torch.compile()
def _aux_update_many_handle_single_chunk(
update_key_chunk: torch.Tensor,
update_value_chunk: torch.Tensor,
eviction_info_chunk: torch.Tensor,
block_table: torch.Tensor,
key_blocks: torch.Tensor,
value_blocks: torch.Tensor,
cache_seq_lenghts: torch.Tensor,
recent_info: torch.Tensor,
recent_info_position: torch.Tensor,
page_batch_index: torch.Tensor,
update_mask: torch.Tensor,
true_update_size: torch.Tensor,
) -> torch.Tensor:
"""
Used for prefilling the KV cache as each tensor has a fixed size.
`true_update_size` represents the true number of elements to be added for each batch index.
"""
assert update_key_chunk.size() == update_value_chunk.size()
page_batch, chunk_len, head_dim = update_key_chunk.size()
assert chunk_len < recent_info.size(1)
assert eviction_info_chunk.size() == (page_batch, chunk_len)
assert page_batch_index.size() == (page_batch,)
block_size = key_blocks.size(1)
device = update_key_chunk.device
# First we update the eviction info for the previous element if present
update_eviction_info_positions = (recent_info_position - 1) % recent_info.size(1)
update_eviction_info_mask = (cache_seq_lenghts > 0).to(torch.int32)
recent_info[page_batch_index, update_eviction_info_positions, 1] = (
eviction_info_chunk[:, 0] * update_eviction_info_mask
+ (1 - update_eviction_info_mask) * recent_info[page_batch_index, update_eviction_info_positions, 1]
).to(torch.int32)
chunk_indexer = torch.arange(chunk_len, dtype=torch.int32, device=device)
# The following trick handles variable lens: if the index is longer than true_update_size, then pad the index
# with the last element within the true_update_size, e.g., [0, 1, 2, 3, 4, 5] and true_update_size = [3]
# means that we have [0, 1, 2, 2, 2, 2] . This will later be used to write the same element multiple times
# while preserving the constant shapes of the tensors.
potential_eviction_positions_in_recent_info = (
recent_info_position[:, None] + torch.minimum(chunk_indexer[None, :], true_update_size[:, None] - 1)
) % recent_info.size(1)
potential_eviction_positions_in_seq = recent_info[
page_batch_index[:, None], potential_eviction_positions_in_recent_info, 0
]
confirmed_evictions_mask = (
recent_info[page_batch_index[:, None], potential_eviction_positions_in_recent_info, 1] == 1
)
# Account for the padding with the last element (as described above) to get a proper count of confirmed evictions
confirmed_evictions_mask[:, 1:] = torch.logical_and(
confirmed_evictions_mask[:, 1:],
potential_eviction_positions_in_recent_info[:, 1:] != potential_eviction_positions_in_recent_info[:, :-1],
)
confirmed_evictions_cum_sum = confirmed_evictions_mask.to(torch.int32).cumsum(dim=-1)
confirmed_evictions_mask = torch.logical_and(
confirmed_evictions_mask,
confirmed_evictions_cum_sum <= true_update_size[:, None],
)
# Count how many new positions are needed for each element of the batch
num_confirmed_evictions = confirmed_evictions_mask.to(torch.int32).sum(dim=-1)
new_positions_used = true_update_size - num_confirmed_evictions
assert (new_positions_used >= 0).all()
assert new_positions_used.size() == (page_batch,)
new_free_positions = cache_seq_lenghts[:, None] + torch.clamp(
torch.minimum(chunk_indexer[None, :], new_positions_used[:, None] - 1), min=0
)
assert new_free_positions.size() == (page_batch, chunk_len)
assert new_free_positions.size() == potential_eviction_positions_in_seq.size()
potential_eviction_positions_in_seq = torch.cat(
[
potential_eviction_positions_in_seq,
new_free_positions,
],
dim=-1,
)
# Padding below allows for constant shape ops to take prefix of length new_positions_used from new_free_positions
confirmed_evictions_padding = torch.zeros_like(confirmed_evictions_mask)
padding_chunk_size = chunk_len - num_confirmed_evictions[:, None]
indexer = torch.minimum(chunk_indexer[None, :], torch.clamp(padding_chunk_size - 1, min=0))
confirmed_evictions_padding[page_batch_index[:, None], indexer] = True
# If only post eviction positions are used, then have writing padding that ends in the last of those positions,
# instead of the next free position
confirmed_evictions_padding = torch.logical_and(confirmed_evictions_padding, padding_chunk_size > 0)
confirmed_evictions_mask = torch.cat([confirmed_evictions_mask, confirmed_evictions_padding], dim=-1)
pad_selector = (new_positions_used > 0).to(torch.int32)[:, None]
potential_eviction_positions_in_seq[:, chunk_len:] = (
pad_selector * potential_eviction_positions_in_seq[:, chunk_len:]
+ (1 - pad_selector) * potential_eviction_positions_in_seq[:, [chunk_len - 1]]
)
new_write_positions = potential_eviction_positions_in_seq[confirmed_evictions_mask].reshape(page_batch, chunk_len)
_aux_write_kv(
block_table=block_table,
blocks=key_blocks,
write_positions=new_write_positions,
values=update_key_chunk,
page_batch_index=page_batch_index,
)
_aux_write_kv(
block_table=block_table,
blocks=value_blocks,
write_positions=new_write_positions,
values=update_value_chunk,
page_batch_index=page_batch_index,
)
recent_indexer = torch.minimum(chunk_indexer[None, :], torch.clamp(true_update_size[:, None] - 1, min=0))
recent_info_indexer = (recent_info_position[:, None] + recent_indexer) % recent_info.size(1)
# Update the info about last window positions
non_empty_update = (true_update_size[:, None] > 0).to(torch.int32)
recent_info[page_batch_index[:, None], recent_info_indexer, 0] = (
new_write_positions * non_empty_update
+ recent_info[page_batch_index[:, None], recent_info_indexer, 0] * (1 - non_empty_update)
).to(torch.int32)
eviction_info_chunk = torch.cat(
[
eviction_info_chunk[:, 1:],
torch.zeros_like(eviction_info_chunk[:, [0]]),
],
dim=-1,
)
recent_info[page_batch_index[:, None], recent_info_indexer, 1] = (
eviction_info_chunk[:, :] * non_empty_update
+ recent_info[page_batch_index[:, None], recent_info_indexer, 1] * (1 - non_empty_update)
).to(torch.int32)
recent_info_position[...] += true_update_size
cache_seq_lenghts[...] += new_positions_used
require_free_pages = torch.logical_and(new_positions_used > 0, cache_seq_lenghts % block_size == 0)
return require_free_pages
class DMSCache(Cache):
def __init__(
self,
dms_window_size: int,
max_context_length: int,
offloading: bool = False,
offload_only_non_sliding: bool = False,
accomodate_min_initial_context_length: int = 2048,
block_size: int = 256,
):
super().__init__(
layer_class_to_replicate=functools.partial(
DMSPagedCacheLayer,
dms_window_size=dms_window_size,
max_context_length=max_context_length,
accomodate_min_initial_context_length=accomodate_min_initial_context_length,
block_size=block_size,
),
offloading=offloading,
offload_only_non_sliding=offload_only_non_sliding,
)
def to_legacy_cache(self):
raise NotImplementedError("Not Supported")
@classmethod
def from_legacy_cache(cls, *args, **kwargs):
raise NotImplementedError("Not Supported")
def early_initialization(self, *args, **kwargs):
raise NotImplementedError("Not Supported")
def __iter__(self):
raise NotImplementedError("Not Supported")
def __getitem__(self, layer_idx: int):
assert layer_idx < len(self.layers)
return self.layers[layer_idx]
class DMSPagedCacheLayer(CacheLayerMixin):
def __init__(
self,
dms_window_size: int,
max_context_length: int,
block_size: int = 256,
growth_factor: float = 1.5,
accomodate_min_initial_context_length: int = 4096,
):
super().__init__()
assert block_size <= dms_window_size
self.block_size = block_size
self.dms_window_size = dms_window_size
self.prefill_chunk_size = max(self.dms_window_size - 2, block_size)
assert self.prefill_chunk_size > 0
self.growth_factor = growth_factor
self.min_initial_context_length = accomodate_min_initial_context_length
self.max_context_length = max_context_length
self.max_blocks_per_sequence = ceil_int_div(self.max_context_length, self.block_size)
self.key_blocks = None
self.value_blocks = None
self.block_table = None
self.free_page_ids = None
self.cache_seq_lengths = None
self.recent_info = None # Position and eviction info of last window_size keys/values
self.recent_info_position = None
self.device = None
self.cumulative_length = 0
def offload(self):
if self.key_blocks is not None:
self.key_blocks = self.key_blocks.to("cpu", non_blocking=True)
self.value_blocks = self.value_blocks.to("cpu", non_blocking=True)
self.block_table = self.block_table.to("cpu", non_blocking=True)
self.free_page_ids = self.free_page_ids.to("cpu", non_blocking=True)
self.cache_seq_lengths = self.cache_seq_lengths.to("cpu", non_blocking=True)
self.recent_info = self.recent_info.to("cpu", non_blocking=True)
self.recent_info_position = self.recent_info_position.to("cpu", non_blocking=True)
def prefetch(self):
if self.key_blocks is not None and self.key_blocks.device != self.device:
self.key_blocks = self.key_blocks.to(self.device, non_blocking=True)
self.value_blocks = self.value_blocks.to(self.device, non_blocking=True)
self.block_table = self.block_table.to(self.device, non_blocking=True)
self.free_page_ids = self.free_page_ids.to(self.device, non_blocking=True)
self.cache_seq_lengths = self.cache_seq_lengths.to(self.device, non_blocking=True)
self.recent_info = self.recent_info.to(self.device, non_blocking=True)
self.recent_info_position = self.recent_info_position.to(self.device, non_blocking=True)
def reset(self) -> None:
"""Resets the cache values while preserving the objects"""
print(f"reset {self.key_blocks is not None}")
if self.key_blocks is not None:
self.key_blocks = None
self.value_blocks = None
self.block_table = None
self.free_page_ids = None
self.cache_seq_lengths = None
self.recent_info = None
self.recent_info_position = None
gc.collect()
torch.cuda.empty_cache()
self.cumulative_length = 0
def reorder_cache(self, beam_idx: torch.LongTensor) -> None:
"""Reorders this layer's cache for beam search."""
assert False # No support for beam search at this point
def _get_free_pages(self, num_pages: int):
while len(self.free_page_ids) < num_pages:
def expand_blocks(blocks: torch.Tensor):
return torch.cat(
[
blocks,
blocks.new_zeros(
(
float_ceil(blocks.size(0) * self.growth_factor) - blocks.size(0),
blocks.size(1),
blocks.size(2),
blocks.size(3),
)
),
],
dim=0,
)
old_num_blocks = self.key_blocks.size(0)
self.key_blocks = expand_blocks(self.key_blocks)
self.value_blocks = expand_blocks(self.value_blocks)
assert self.key_blocks.size(0) == self.value_blocks.size(0)
self.free_page_ids = torch.cat(
[
self.free_page_ids,
torch.arange(
old_num_blocks,
self.key_blocks.size(0),
dtype=torch.int32,
device=self.device,
),
],
dim=0,
)
result = self.free_page_ids[:num_pages]
assert result.size() == (num_pages,)
self.free_page_ids = self.free_page_ids[num_pages:]
return result
def lazy_initialization(self, key_states: torch.Tensor):
self.dtype, self.device = key_states.dtype, key_states.device
self.batch_size, self.num_heads, _, self.head_dim = key_states.shape
self.page_batch = self.batch_size * self.num_heads
initial_num_blocks = max(
ceil_int_div(self.min_initial_context_length, self.block_size) * self.page_batch,
self.page_batch,
)
self.block_table = -torch.ones(
self.page_batch,
self.max_blocks_per_sequence + 1, # +1 for handling full cache case
dtype=torch.int32,
device=self.device,
)
self.key_blocks = torch.zeros(
(initial_num_blocks, self.block_size, 1, self.head_dim),
dtype=self.dtype,
device=self.device,
)
self.value_blocks = torch.zeros(
(initial_num_blocks, self.block_size, 1, self.head_dim),
dtype=self.dtype,
device=self.device,
)
self.free_page_ids = torch.arange(0, initial_num_blocks, dtype=torch.int32, device=self.device)
self.cache_seq_lengths = torch.zeros(self.page_batch, dtype=torch.int32, device=self.device)
self.recent_info = torch.zeros(
(self.page_batch, self.dms_window_size, 2),
dtype=torch.int32,
device=self.device,
)
self.recent_info_position = torch.zeros((self.page_batch,), dtype=torch.int32, device=self.device)
self.block_table[:, 0] = self._get_free_pages(self.block_table.size(0))
def _handle_page_allocation(self, requires_free_page: torch.Tensor, page_batch_index: torch.Tensor):
if requires_free_page.any():
req_free_pages = page_batch_index[requires_free_page]
free_pages = self._get_free_pages(len(req_free_pages))
self.block_table[
req_free_pages,
self.cache_seq_lengths[req_free_pages] // self.block_size,
] = free_pages
def _update_single(
self,
key_states: torch.Tensor,
value_states: torch.Tensor,
eviction_info: torch.Tensor,
):
batch_x_head, seq_len, head_dim = key_states.size()
page_batch_index = torch.arange(batch_x_head, dtype=torch.int32, device=self.device)
assert seq_len == 1
requires_free_page = _aux_update_single(
key_states=key_states,
value_states=value_states,
eviction_info=eviction_info,
recent_info=self.recent_info,
recent_info_position=self.recent_info_position,
block_table=self.block_table,
key_blocks=self.key_blocks,
value_blocks=self.value_blocks,
cache_seq_lenghts=self.cache_seq_lengths,
page_batch_index=page_batch_index,
)
self._handle_page_allocation(requires_free_page=requires_free_page, page_batch_index=page_batch_index)
# NOTE: Prefill is not yet optimized
def _update_many(
self,
key_states: torch.Tensor,
value_states: torch.Tensor,
eviction_info: torch.Tensor,
sequence_lengths: torch.Tensor,
):
# Assume key and value states are left padded, e.g., [_, _, _, 1, 2, 3, 4]
page_batch, seq_len, head_dim = key_states.size()
assert page_batch == self.page_batch
assert head_dim == self.head_dim
assert eviction_info.size() == (page_batch, seq_len)
assert sequence_lengths.ndim == 1
assert sequence_lengths.min() > 0, sequence_lengths
start_positions = seq_len - sequence_lengths
end_positions = start_positions + sequence_lengths
page_batch_index = torch.arange(page_batch, dtype=torch.int32, device=self.device)
while (start_positions < end_positions).any():
chunk_indexer = torch.arange(self.prefill_chunk_size, dtype=torch.int32, device=self.device)[None, :]
update_mask = chunk_indexer < (self.block_size - (self.cache_seq_lengths[:, None] % self.block_size))
chunk_indexer = start_positions[:, None] + chunk_indexer
update_mask = torch.logical_and(update_mask, chunk_indexer < end_positions[:, None])
chunk_indexer = torch.clamp(torch.minimum(chunk_indexer, end_positions[:, None] - 1), min=0)
true_update_size = update_mask.to(torch.int32).sum(dim=1)
chunk_indexer = torch.clamp(
torch.minimum(
chunk_indexer,
start_positions[:, None] + true_update_size[:, None] - 1,
),
min=0,
)
key_chunk = key_states[page_batch_index[:, None], chunk_indexer]
value_chunk = value_states[page_batch_index[:, None], chunk_indexer]
eviction_info_chunk = eviction_info[page_batch_index[:, None], chunk_indexer]
requires_free_page = _aux_update_many_handle_single_chunk(
update_key_chunk=key_chunk,
update_value_chunk=value_chunk,
eviction_info_chunk=eviction_info_chunk,
block_table=self.block_table,
key_blocks=self.key_blocks,
value_blocks=self.value_blocks,
cache_seq_lenghts=self.cache_seq_lengths,
recent_info=self.recent_info,
recent_info_position=self.recent_info_position,
page_batch_index=page_batch_index,
update_mask=update_mask,
true_update_size=true_update_size,
)
self._handle_page_allocation(requires_free_page=requires_free_page, page_batch_index=page_batch_index)
start_positions[...] += true_update_size
def get_contiguous_cache(
self, right_padded: bool = True
) -> tuple[torch.Tensor, torch.Tensor, torch.Tensor, torch.Tensor]:
assert self.key_blocks is not None
num_blocks_per_sequence = self.cache_seq_lengths.max().item() // self.block_size + 1
blocks_to_retrieve = self.block_table[:, :num_blocks_per_sequence]
# page_batch_index = torch.arange(self.batch_size * self.num_heads, dtype=torch.int32, device=self.device)[:, None]
max_length = self.cache_seq_lengths.max().item()
def handle_one(
blocks: torch.Tensor,
blocks_to_retrieve: torch.Tensor,
max_length: int,
num_blocks_per_sequence: int,
):
retrieved = blocks[blocks_to_retrieve].reshape(
self.page_batch,
num_blocks_per_sequence * self.block_size,
self.head_dim,
)
retrieved = retrieved[:, :max_length, :]
recent_info_size = torch.clamp(self.cache_seq_lengths, max=self.dms_window_size)
window_index = torch.arange(self.dms_window_size, device=self.device, dtype=torch.int32)
adjusted_window_index = torch.minimum(
window_index[None, :], torch.clamp(recent_info_size[:, None] - 1, min=0)
)
last_pos_data_ptr = (self.recent_info_position[:, None] - 1 - adjusted_window_index) % self.dms_window_size
page_batch_index = torch.arange(self.page_batch, device=self.device, dtype=torch.int32)
window_positions = self.recent_info[page_batch_index[:, None], last_pos_data_ptr, 0]
permutation_index = self.cache_seq_lengths[:, None] - 1 - adjusted_window_index
assert (permutation_index >= 0).all()
permutation = torch.arange(max_length, device=blocks.device, dtype=torch.int32)[None, :]
permutation = torch.minimum(permutation, self.cache_seq_lengths[:, None] - 1)
permutation = torch.broadcast_to(permutation, (self.page_batch, max_length))
non_window_positions = permutation[:, :, None] != window_positions[:, None, :]
non_window_positions = non_window_positions.to(torch.int32).min(dim=-1).values.to(torch.bool)
result_permutation = torch.zeros_like(permutation)
result_permutation[page_batch_index[:, None], permutation_index] = window_positions
# Not yet optimized
num_non_window_positions = non_window_positions.to(torch.int32).sum(dim=-1).cpu().tolist()
for i in range(self.page_batch):
result_permutation[i, : num_non_window_positions[i]] = permutation[i, non_window_positions[i]]
return retrieved[page_batch_index[:, None], result_permutation]
retrieved_keys = handle_one(
blocks=self.key_blocks,
blocks_to_retrieve=blocks_to_retrieve,
max_length=max_length,
num_blocks_per_sequence=num_blocks_per_sequence,
)
retrieved_values = handle_one(
blocks=self.value_blocks,
blocks_to_retrieve=blocks_to_retrieve,
max_length=max_length,
num_blocks_per_sequence=num_blocks_per_sequence,
)
cache_seq_lenghts = self.cache_seq_lengths
page_batch_index = torch.arange(self.page_batch, device=self.device, dtype=torch.int32)
eviction_info_indexer = torch.arange(self.dms_window_size, device=self.device, dtype=torch.int32)
eviction_info_indexer = torch.minimum(
eviction_info_indexer[None, :],
_aux_get_recent_position_size(
cache_seq_lenghts=self.cache_seq_lengths,
dms_window_size=self.dms_window_size,
)[:, None]
- 1,
)
eviction_info_indexer = (
_aux_get_first_recent_position(
recent_info_position=self.recent_info_position,
cache_seq_lenghts=self.cache_seq_lengths,
dms_window_size=self.dms_window_size,
)[:, None]
+ eviction_info_indexer
)
eviction_info_indexer = eviction_info_indexer % self.dms_window_size
eviction_info = self.recent_info[page_batch_index[:, None], eviction_info_indexer, 1]
if not right_padded:
# Slow, but only used only in prefill
def left_pad_one(x: torch.Tensor, lens: torch.Tensor):
max_length = x.shape[1]
lens = lens.cpu().tolist()
left_padded_content = []
for i in range(self.page_batch):
content = x[i, : lens[i]]
content = torch.cat(
[
x.new_zeros(
(
max_length - lens[i],
*content.shape[1:],
)
),
content,
],
dim=0,
)
left_padded_content.append(content)
return torch.stack(left_padded_content, dim=0)
retrieved_keys = left_pad_one(retrieved_keys, cache_seq_lenghts)
retrieved_values = left_pad_one(retrieved_values, cache_seq_lenghts)
cache_seq_lenghts = cache_seq_lenghts.reshape(self.batch_size, self.num_heads)
eviction_info = left_pad_one(
eviction_info,
_aux_get_recent_position_size(
cache_seq_lenghts=self.cache_seq_lengths,
dms_window_size=self.dms_window_size,
),
)
retrieved_keys = retrieved_keys.reshape(
self.batch_size, self.num_heads, max_length, self.head_dim
).contiguous()
retrieved_values = retrieved_values.reshape(
self.batch_size, self.num_heads, max_length, self.head_dim
).contiguous()
cache_seq_lenghts = cache_seq_lenghts.reshape(self.batch_size, self.num_heads)
eviction_info = eviction_info.reshape(self.batch_size, self.num_heads, -1)
return retrieved_keys, retrieved_values, cache_seq_lenghts, eviction_info
def update(
self,
key_states: torch.Tensor,
value_states: torch.Tensor,
cache_kwargs: dict[str, Any],
):
eviction_info = cache_kwargs["eviction_info"]
sequence_lengths = cache_kwargs["sequence_lengths"]
cumulative_length = cache_kwargs["cumulative_length"]
if self.key_blocks is None:
self.lazy_initialization(key_states)
batch, head, seq_len, head_dim = key_states.size()
assert key_states.size() == value_states.size()
assert key_states.size()[:3] == eviction_info.size()
assert sequence_lengths is None or sequence_lengths.size() == (batch, head)
assert batch * head == self.page_batch
assert self.head_dim == head_dim
key_states = key_states.reshape(self.page_batch, seq_len, head_dim)
value_states = value_states.reshape(self.page_batch, seq_len, head_dim)
eviction_info = eviction_info.reshape(self.page_batch, seq_len)
if sequence_lengths is not None:
sequence_lengths = sequence_lengths.reshape(self.page_batch)
if seq_len == 1:
assert sequence_lengths is None or (sequence_lengths == 1).all()
assert cumulative_length == 1
self._update_single(
key_states=key_states,
value_states=value_states,
eviction_info=eviction_info,
)
else:
self._update_many(
key_states=key_states,
value_states=value_states,
eviction_info=eviction_info,
sequence_lengths=sequence_lengths,
)
self.cumulative_length += cumulative_length
return None, None
def get_block_table(self):
return self.block_table
def get_key_blocks(self):
return self.key_blocks
def get_value_blocks(self):
return self.value_blocks
def get_seq_lengths(self):
return self.cache_seq_lengths
def get_mask_sizes(self, cache_position: torch.Tensor) -> tuple[int, int]:
"""Returns the length and offset of the cache, used to generate the mask."""
kv_offset = 0
query_length = cache_position.shape[0]
past_seen_tokens = self.get_seq_length()
kv_length = query_length + past_seen_tokens
return kv_length, kv_offset
def get_seq_length(self) -> int:
"""Returns the sequence length of the cached states."""
return self.cumulative_length
def get_max_cache_shape(self) -> int:
"""Returns the maximum sequence length of the cache object. DynamicLayer does not have a maximum length."""
return self.max_context_length
|