Co-authored-by: Xuchun Shang <xuchun.shang@gmail.com> Co-authored-by: ybyang <10629930+whybeyoung@users.noreply.github.com> Co-authored-by: Shangming Cai <csmthu@gmail.com>
290 lines
8.6 KiB
Python
290 lines
8.6 KiB
Python
import hashlib
|
|
import logging
|
|
import os
|
|
from abc import ABC, abstractmethod
|
|
from dataclasses import dataclass
|
|
from typing import Any, List, Optional
|
|
|
|
import torch
|
|
|
|
from sglang.srt.mem_cache.memory_pool_host import HostKVCache
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def get_hash_str(token_ids: List[int], prior_hash: str = None) -> str:
|
|
hasher = hashlib.sha256()
|
|
|
|
if prior_hash:
|
|
hasher.update(bytes.fromhex(prior_hash))
|
|
|
|
for t in token_ids:
|
|
if isinstance(t, tuple):
|
|
# EAGLE bigram mode: hash both elements to uniquely identify the bigram
|
|
for elem in t:
|
|
hasher.update(elem.to_bytes(4, byteorder="little", signed=False))
|
|
else:
|
|
# Regular mode: single integer token
|
|
hasher.update(t.to_bytes(4, byteorder="little", signed=False))
|
|
|
|
return hasher.hexdigest()
|
|
|
|
|
|
def hash_str_to_int64(hash_str: str) -> int:
|
|
"""Convert SHA256 hex string to signed 64-bit integer for events.
|
|
|
|
Takes first 16 hex characters (64 bits) and converts to signed int64 range.
|
|
"""
|
|
# Take first 16 hex chars to get 64-bit value
|
|
uint64_val = int(hash_str[:16], 16)
|
|
# Convert to signed int64 range [-2^63, 2^63-1]
|
|
if uint64_val >= 2**63:
|
|
return uint64_val - 2**64
|
|
return uint64_val
|
|
|
|
|
|
@dataclass
|
|
class HiCacheStorageConfig:
|
|
tp_rank: int
|
|
tp_size: int
|
|
pp_rank: int
|
|
pp_size: int
|
|
is_mla_model: bool
|
|
is_page_first_layout: bool
|
|
model_name: Optional[str]
|
|
extra_config: Optional[dict] = None
|
|
|
|
|
|
@dataclass
|
|
class HiCacheStorageExtraInfo:
|
|
prefix_keys: Optional[List[str]] = (None,)
|
|
extra_info: Optional[dict] = None
|
|
|
|
|
|
class HiCacheStorage(ABC):
|
|
"""
|
|
HiCacheStorage is a class that provides a generic key-value interface for storing and retrieving KV cache.
|
|
It abstracts the underlying storage mechanism, allowing different implementations to be used.
|
|
"""
|
|
|
|
# todo, the page size of storage backend does not have to be the same as the same as host memory pool
|
|
|
|
def register_mem_pool_host(self, mem_pool_host: HostKVCache):
|
|
self.mem_pool_host = mem_pool_host
|
|
|
|
def batch_get_v1(
|
|
self,
|
|
keys: List[str],
|
|
host_indices: torch.Tensor,
|
|
extra_info: Optional[HiCacheStorageExtraInfo] = None,
|
|
) -> List[bool]:
|
|
"""
|
|
Retrieve values for multiple keys.
|
|
Returns a list of booleans indicating success for each key.
|
|
"""
|
|
pass
|
|
|
|
def batch_set_v1(
|
|
self,
|
|
keys: List[str],
|
|
host_indices: torch.Tensor,
|
|
extra_info: Optional[HiCacheStorageExtraInfo] = None,
|
|
) -> List[bool]:
|
|
"""
|
|
Store multiple key-value pairs.
|
|
Returns a list of booleans indicating success for each key.
|
|
"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def get(
|
|
self,
|
|
key: str,
|
|
target_location: Optional[Any] = None,
|
|
target_sizes: Optional[Any] = None,
|
|
) -> torch.Tensor | None:
|
|
"""
|
|
Retrieve the value associated with the given key.
|
|
Returns None if the key does not exist.
|
|
"""
|
|
pass
|
|
|
|
# TODO: Deprecate
|
|
@abstractmethod
|
|
def batch_get(
|
|
self,
|
|
keys: List[str],
|
|
target_locations: Optional[Any] = None,
|
|
target_sizes: Optional[Any] = None,
|
|
) -> List[torch.Tensor | None] | int:
|
|
"""
|
|
Retrieve values for multiple keys.
|
|
Returns a list of tensors or None for each key.
|
|
"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def set(
|
|
self,
|
|
key: str,
|
|
value: Optional[Any] = None,
|
|
target_location: Optional[Any] = None,
|
|
target_sizes: Optional[Any] = None,
|
|
) -> bool:
|
|
"""
|
|
Store the value associated with the given key.
|
|
Returns True if the operation was successful, False otherwise.
|
|
"""
|
|
pass
|
|
|
|
# TODO: Deprecate
|
|
@abstractmethod
|
|
def batch_set(
|
|
self,
|
|
keys: List[str],
|
|
values: Optional[Any] = None,
|
|
target_locations: Optional[Any] = None,
|
|
target_sizes: Optional[Any] = None,
|
|
) -> bool:
|
|
"""
|
|
Store multiple key-value pairs.
|
|
Returns True if all operations were successful, False otherwise.
|
|
"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def exists(self, key: str) -> bool:
|
|
"""
|
|
Check if the key exists in the storage.
|
|
Returns True if the key exists, False otherwise.
|
|
"""
|
|
pass
|
|
|
|
# TODO: Use a finer-grained return type (e.g., List[bool])
|
|
def batch_exists(
|
|
self, keys: List[str], extra_info: Optional[HiCacheStorageExtraInfo] = None
|
|
) -> int:
|
|
"""
|
|
Check if the keys exist in the storage.
|
|
return the number of consecutive existing keys from the start.
|
|
Can be overridden by subclasses for more efficient implementation.
|
|
"""
|
|
for i in range(len(keys)):
|
|
if not self.exists(keys[i]):
|
|
return i
|
|
return len(keys)
|
|
|
|
def clear(self) -> None:
|
|
pass
|
|
|
|
def get_stats(self):
|
|
return None
|
|
|
|
|
|
class HiCacheFile(HiCacheStorage):
|
|
|
|
def __init__(
|
|
self, storage_config: HiCacheStorageConfig, file_path: str = "/tmp/hicache"
|
|
):
|
|
self.file_path = os.getenv("SGLANG_HICACHE_FILE_BACKEND_STORAGE_DIR", file_path)
|
|
|
|
tp_rank, tp_size, model_name, is_mla_model = (
|
|
storage_config.tp_rank,
|
|
storage_config.tp_size,
|
|
storage_config.model_name,
|
|
storage_config.is_mla_model,
|
|
)
|
|
model_name = "-".join(model_name.split("/")) if model_name else ""
|
|
if is_mla_model:
|
|
self.config_suffix = f"_{model_name}"
|
|
else:
|
|
self.config_suffix = f"_{model_name}_{tp_rank}_{tp_size}"
|
|
|
|
if not os.path.exists(self.file_path) and tp_rank == 0:
|
|
os.makedirs(self.file_path)
|
|
logger.info(f"Created HiCacheFile storage directory at {self.file_path}")
|
|
|
|
def _get_suffixed_key(self, key: str) -> str:
|
|
return key + self.config_suffix
|
|
|
|
def get(
|
|
self,
|
|
key: str,
|
|
target_location: torch.Tensor,
|
|
target_sizes: Optional[Any] = None,
|
|
) -> torch.Tensor | None:
|
|
key = self._get_suffixed_key(key)
|
|
tensor_path = os.path.join(self.file_path, f"{key}.bin")
|
|
try:
|
|
expected = target_location.numel() * target_location.element_size()
|
|
with open(tensor_path, "rb", buffering=0) as f:
|
|
buf = memoryview(target_location.view(torch.uint8).contiguous().numpy())
|
|
if f.readinto(buf) != expected:
|
|
raise IOError(f"Short read for {key}")
|
|
return target_location
|
|
except FileNotFoundError:
|
|
logger.warning(f"Failed to fetch {key} from HiCacheFile storage.")
|
|
return None
|
|
|
|
def batch_get(
|
|
self,
|
|
keys: List[str],
|
|
target_locations: List[torch.Tensor],
|
|
target_sizes: Optional[Any] = None,
|
|
) -> List[torch.Tensor | None]:
|
|
return [
|
|
self.get(key, target_location)
|
|
for key, target_location in zip(
|
|
keys, target_locations or [None] * len(keys)
|
|
)
|
|
]
|
|
|
|
def set(
|
|
self,
|
|
key: str,
|
|
value: Optional[Any] = None,
|
|
target_location: Optional[Any] = None,
|
|
target_sizes: Optional[Any] = None,
|
|
) -> bool:
|
|
if self.exists(key):
|
|
logger.debug(f"Key {key} already exists. Skipped.")
|
|
return True
|
|
|
|
key = self._get_suffixed_key(key)
|
|
tensor_path = os.path.join(self.file_path, f"{key}.bin")
|
|
try:
|
|
value.contiguous().view(dtype=torch.uint8).numpy().tofile(tensor_path)
|
|
return True
|
|
except Exception as e:
|
|
logger.error(f"Failed to save tensor {key}: {e}")
|
|
return False
|
|
|
|
def batch_set(
|
|
self,
|
|
keys: List[str],
|
|
values: Optional[Any] = None,
|
|
target_locations: Optional[Any] = None,
|
|
target_sizes: Optional[Any] = None,
|
|
) -> bool:
|
|
for key, value in zip(keys, values):
|
|
if not self.set(key, value):
|
|
return False
|
|
return True
|
|
|
|
def exists(self, key: str) -> bool:
|
|
key = self._get_suffixed_key(key)
|
|
tensor_path = os.path.join(self.file_path, f"{key}.bin")
|
|
return os.path.exists(tensor_path)
|
|
|
|
def clear(self) -> bool:
|
|
try:
|
|
for filename in os.listdir(self.file_path):
|
|
file_path = os.path.join(self.file_path, filename)
|
|
if os.path.isfile(file_path):
|
|
os.remove(file_path)
|
|
logger.info("Cleared all entries in HiCacheFile storage.")
|
|
return True
|
|
except Exception as e:
|
|
logger.error(f"Failed to clear HiCacheFile storage: {e}")
|
|
return False
|