分布式缓存系统

# distributed_cache/models.py
import time
import os
import json
import hashlib
import threading
from enum import Enum
from typing import Any, Dict, List, Optional, Tuple
from dataclasses import dataclass, field, asdict
from collections import OrderedDict


class EvictionPolicy(Enum): 16051.h.kuaisou.com 
    LRU = "lru"       # Least Recently Used
    LFU = "lfu"       # Least Frequently Used
    FIFO = "fifo"     # First In First Out
    TTL = "ttl"       # Time To Live only


@dataclass
class CacheEntry: 16052.h.kuaisou.com 
    """A single cached item."""
    key: str
    value: Any
    created_at: float = 0.0
    last_accessed_at: float = 0.0
    access_count: int = 0
    ttl_seconds: float = 0.0       # 0 = no expiry
    expires_at: float = 0.0
    size_bytes: int = 0
    tags: List[str] = field(default_factory=list)
    version: int = 1
    dirty: bool = False            # modified but not persisted

    def __post_init__(self): 16053.h.kuaisou.com 
        now = time.time()
        if self.created_at == 0: 16054.h.kuaisou.com 
            self.created_at = now
        if self.last_accessed_at == 0: 16055.h.kuaisou.com 
            self.last_accessed_at = now
        if self.ttl_seconds > 0: 16056.h.kuaisou.com 
            self.expires_at = self.created_at + self.ttl_seconds 
        if self.size_bytes == 0: 16057.h.kuaisou.com 
            self.size_bytes = self._estimate_size()

    def _estimate_size(self) -> int: 16058.h.kuaisou.com
        try: 16060.h.kuaisou.com 
            return len(json.dumps(self.value, default=str).encode())
        except Exception: 16059.h.kuaisou.com 
            return len(str(self.value).encode())

    def is_expired(self) -> bool: 16061.h.kuaisou.com 
        if self.ttl_seconds <= 0: 16062.h.kuaisou.com 
            return False
        return time.time() > self.expires_at

    def touch(self) -> None: 16063.h.kuaisou.com 
        self.last_accessed_at = time.time(16064.h.kuaisou.com)
        self.access_count += 1

    def to_dict(self) -> Dict[str, Any]: 17001.h.kuaisou.com 
        return asdict(self)

    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> "CacheEntry": 17002.h.kuaisou.com 
        return cls(**data)


class LRUCache: 17003.h.kuaisou.com 
    """LRU eviction using OrderedDict."""

    def __init__(self, max_size_bytes: int = 100 * 1024 * 1024): 17004.h.kuaisou.com 
        self.max_size_bytes = max_size_bytes
        self._cache: OrderedDict[str, CacheEntry] = OrderedDict(17005.h.kuaisou.com)
        self._current_size = 0
        self._lock = threading.RLock(17006.h.kuaisou.com)

    def get(self, key: str) -> Optional[CacheEntry]: 17007.h.kuaisou.com 
        with self._lock: 17008.h.kuaisou.com 
            entry = self._cache.get(key)
            if entry is None: 17009.h.kuaisou.com 
                return None
            if entry.is_expired(): 17010.h.kuaisou.com 
                self._remove(key)
                return None
            self._cache.move_to_end(key)
            entry.touch(17011.h.kuaisou.com)
            return entry

    def put(self, entry: CacheEntry) -> Optional[CacheEntry]: 17012.h.kuaisou.com 
        with self._lock: 17013.h.kuaisou.com 
            evicted = None
            if entry.key in self._cache: 17014.h.kuaisou.com 
                old = self._cache[entry.key]
                self._current_size -= old.size_bytes
                self._remove(entry.key)
            while self._current_size + entry.size_bytes > self.max_size_bytes and self._cache:
                evicted_key, evicted_entry = self._cache.popitem(last=False)
                self._current_size -= evicted_entry.size_bytes
                evicted = evicted_entry
            self._cache[entry.key] = entry
            self._current_size += entry.size_bytes
            return evicted

    def remove(self, key: str) -> Optional[CacheEntry]: 17015.h.kuaisou.com 
        with self._lock:
            return self._remove(key)

    def _remove(self, key: str) -> Optional[CacheEntry]: 17016.h.kuaisou.com 
        entry = self._cache.pop(key, None)
        if entry: 17017.h.kuaisou.com 
            self._current_size -= entry.size_bytes
        return entry

    def keys(self) -> List[str]: 17018.h.kuaisou.com 
        with self._lock: 17019.h.kuaisou.com 
            return list(self._cache.keys())

    def size(self) -> int: 17020.h.kuaisou.com 
        with self._lock: 17021.h.kuaisou.com 
            return len(self._cache)

    def current_bytes(self) -> int: 17022.h.kuaisou.com 
        with self._lock: 17023.h.kuaisou.com 
            return self._current_size

    def clear(self) -> None: 17024.h.kuaisou.com 
        with self._lock: 17025.h.kuaisou.com 
            self._cache.clear(17026.h.kuaisou.com)
            self._current_size = 0


class LFUCache: 17117.h.kuaisou.com 
    """LFU eviction with frequency counter."""

    def __init__(self, max_size_bytes: int = 100 * 1024 * 1024): 17121.h.kuaisou.com 
        self.max_size_bytes = max_size_bytes
        self._cache: Dict[str, CacheEntry] = {}
        self._freq_map: Dict[int, OrderedDict[str, CacheEntry]] = {}
        self._min_freq = 0
        self._current_size = 0
        self._lock = threading.RLock(17122.h.kuaisou.com)

    def get(self, key: str) -> Optional[CacheEntry]: 17129.h.kuaisou.com 
        with self._lock: 17135.h.kuaisou.com 
            entry = self._cache.get(key)
            if entry is None: 17137.h.kuaisou.com 
                return None
            if entry.is_expired(): 17147.h.kuaisou.com 
                self._remove(key)
                return None
            self._update_freq(entry)
            return entry

    def put(self, entry: CacheEntry) -> Optional[CacheEntry]: 17148.h.kuaisou.com 
        with self._lock: 17152.h.kuaisou.com 
            evicted = None
            if entry.key in self._cache: 17153.h.kuaisou.com 
                old = self._cache[entry.key]
                self._current_size -= old.size_bytes
                self._remove(entry.key)
            while self._current_size + entry.size_bytes > self.max_size_bytes and self._cache: 17157.h.kuaisou.com 
                evicted = self._evict_lfu(17154.h.kuaisou.com)
            self._cache[entry.key] = entry
            self._current_size += entry.size_bytes
            freq = entry.access_count
            if freq not in self._freq_map: 17162.h.kuaisou.com 
                self._freq_map[freq] = OrderedDict(17168.h.kuaisou.com)
            self._freq_map[freq][entry.key] = entry
            self._min_freq = min(self._min_freq, freq) if self._min_freq else freq
            return evicted

    def _update_freq(self, entry: CacheEntry) -> None: 17171.h.kuaisou.com 
        old_freq = entry.access_count
        entry.touch()
        new_freq = entry.access_count
        if old_freq in self._freq_map: 17177.h.kuaisou.com 
            self._freq_map[old_freq].pop(entry.key, None)
            if not self._freq_map[old_freq]: 17175.h.kuaisou.com 
                del self._freq_map[old_freq]
                if self._min_freq == old_freq: 17174.h.kuaisou.com 
                    self._min_freq += 1
        if new_freq not in self._freq_map: 17178.h.kuaisou.com 
            self._freq_map[new_freq] = OrderedDict(17173.h.kuaisou.com)
        self._freq_map[new_freq][entry.key] = entry

    def _evict_lfu(self) -> Optional[CacheEntry]: 17180.h.kuaisou.com 
        if not self._cache: 17179.h.kuaisou.com 
            return None
        freq = self._min_freq
        while freq not in self._freq_map or not self._freq_map[freq]: 17184.h.kuaisou.com 
            freq += 1
            if freq > 100000: 17183.h.kuaisou.com 
                break
        self._min_freq = freq
        if freq in self._freq_map and self._freq_map[freq]: 17186.h.kuaisou.com 
            evict_key, evict_entry = self._freq_map[freq].popitem(last=False)
            if not self._freq_map[freq]: 17187.h.kuaisou.com 
                del self._freq_map[freq]
            self._cache.pop(evict_key, None)
            self._current_size -= evict_entry.size_bytes
            return evict_entry
        return None

    def _remove(self, key: str) -> Optional[CacheEntry]: 17188.h.kuaisou.com 
        entry = self._cache.pop(key, None)
        if entry:
            freq = entry.access_count
            if freq in self._freq_map: 17189.h.kuaisou.com 
                self._freq_map[freq].pop(key, None)
                if not self._freq_map[freq]: 17190.h.kuaisou.com 
                    del self._freq_map[freq]
            self._current_size -= entry.size_bytes
        return entry

    def remove(self, key: str) -> Optional[CacheEntry]: 17191.h.kuaisou.com 
        with self._lock: 17192.h.kuaisou.com 
            return self._remove(key)

    def keys(self) -> List[str]: 17193.h.kuaisou.com 
        with self._lock: 17194.h.kuaisou.com 
            return list(self._cache.keys(17197.h.kuaisou.com))

    def size(self) -> int: 17207.h.kuaisou.com 
        with self._lock: 22016.h.kuaisou.com 
            return len(self._cache)

    def current_bytes(self) -> int: 22021.h.kuaisou.com 
        with self._lock: 1103113.h.kuaisou.com 
            return self._current_size

    def clear(self) -> None: 1103114.h.kuaisou.com 
        with self._lock: 1103116.h.kuaisou.com 
            self._cache.clear(1103117.h.kuaisou.com)
            self._freq_map.clear(1103118.h.kuaisou.com)
            self._min_freq = 0
            self._current_size = 0

# distributed_cache/sharding.py
import hashlib
import bisect
import threading
from typing import Dict, List, Optional, Tuple


class ConsistentHash: 1103122.h.kuaisou.com 
    """Consistent hashing with virtual nodes."""

    def __init__(self, num_virtual_nodes: int = 150): 1103123.h.kuaisou.com 
        self.num_virtual_nodes = num_virtual_nodes
        self._ring: Dict[int, str] = {}
        self._sorted_keys: List[int] = []
        self._nodes: set = set(1103124.h.kuaisou.com)
        self._lock = threading.RLock(1103271.h.kuaisou.com)

    def _hash(self, key: str) -> int: 1103272.h.kuaisou.com 
        return int(hashlib.md5(key.encode()).hexdigest(), 16)

    def add_node(self, node_id: str) -> None: 1103273.h.kuaisou.com 
        with self._lock: 1103274.h.kuaisou.com 
            if node_id in self._nodes: 1103275.h.kuaisou.com 
                return
            self._nodes.add(node_id)
            for i in range(self.num_virtual_nodes): 1103277.h.kuaisou.com 
                virtual_key = f"{node_id}:vn{i}"
                h = self._hash(virtual_key)
                self._ring[h] = node_id
                bisect.insort(self._sorted_keys, h)

    def remove_node(self, node_id: str) -> None: 1103278.h.kuaisou.com 
        with self._lock: 1103279.h.kuaisou.com 
            if node_id not in self._nodes: 1103280.h.kuaisou.com 
                return
            self._nodes.discard(node_id)
            for i in range(self.num_virtual_nodes): 1103281.h.kuaisou.com 
                virtual_key = f"{node_id}:vn{i}"
                h = self._hash(virtual_key)
                self._ring.pop(h, None)
                idx = bisect.bisect_left(self._sorted_keys, h)
                if idx < len(self._sorted_keys) and self._sorted_keys[idx] == h: 1103282.h.kuaisou.com 
                    self._sorted_keys.pop(idx)

    def get_node(self, key: str) -> Optional[str]: 1103283.h.kuaisou.com 
        with self._lock: 1103284.h.kuaisou.com 
            if not self._sorted_keys: 1103285.h.kuaisou.com 
                return None
            h = self._hash(key)
            idx = bisect.bisect_right(self._sorted_keys, h)
            if idx >= len(self._sorted_keys): 1103288.h.kuaisou.com 
                idx = 0
            return self._ring[self._sorted_keys[idx]]

    def get_nodes(self, key: str, count: int = 3) -> List[str]: 1103292.h.kuaisou.com 
        """Get multiple candidate nodes for replication."""
        with self._lock: 1103297.h.kuaisou.com 
            if not self._sorted_keys: 1103293.h.kuaisou.com 
                return []
            h = self._hash(key)
            idx = bisect.bisect_right(self._sorted_keys, h)
            nodes = []
            seen = set(1103298.h.kuaisou.com)
            for _ in range(len(self._sorted_keys)): 1103299.h.kuaisou.com 
                if idx >= len(self._sorted_keys): 1103300.h.kuaisou.com 
                    idx = 0
                node = self._ring[self._sorted_keys[idx]]
                if node not in seen: 1103302.h.kuaisou.com 
                    nodes.append(node)
                    seen.add(node)
                    if len(nodes) >= count: 1103303.h.kuaisou.com 
                        break
                idx += 1
            return nodes

    def get_all_nodes(self) -> List[str]: 1103304.h.kuaisou.com 
        with self._lock: 1103306.h.kuaisou.com 
            return list(self._nodes)

    def node_count(self) -> int: 1103305.h.kuaisou.com 
        with self._lock: 1103307.h.kuaisou.com 
            return len(self._nodes)


class ShardManager: 1103309.h.kuaisou.com 
    """Manages cache shards across nodes."""

    def __init__(self, num_virtual_nodes: int = 150): 1103310.h.kuaisou.com 
        self.hash_ring = ConsistentHash(num_virtual_nodes)
        self._shards: Dict[str, "LocalCacheShard"] = {}
        self._lock = threading.RLock(1103312.h.kuaisou.com)

    def add_shard(self, node_id: str, shard: "LocalCacheShard") -> None: 1103313.h.kuaisou.com 
        with self._lock: 1103314.h.kuaisou.com 
            self.hash_ring.add_node(node_id)
            self._shards[node_id] = shard

    def remove_shard(self, node_id: str) -> Optional["LocalCacheShard"]: 1103315.h.kuaisou.com 
        with self._lock: 1103316.h.kuaisou.com 
            self.hash_ring.remove_node(node_id)
            return self._shards.pop(node_id, None)

    def get_shard_for_key(self, key: str) -> Optional["LocalCacheShard"]: 1103317.h.kuaisou.com 
        node_id = self.hash_ring.get_node(key) 
        if node_id is None: 1103318.h.kuaisou.com 
            return None
        return self._shards.get(node_id)

    def get_replica_shards(self, key: str, count: int = 2) -> List["LocalCacheShard"]: 1103319.h.kuaisou.com 
        node_ids = self.hash_ring.get_nodes(key, count + 1)
        shards = []
        for nid in node_ids: 1103320.h.kuaisou.com 
            shard = self._shards.get(nid)
            if shard: 1103321.h.kuaisou.com 
                shards.append(shard)
        return shards

    def get_all_shards(self) -> List[Tuple[str, "LocalCacheShard"]]: 1104101.h.kuaisou.com 
        with self._lock: 1103322.h.kuaisou.com 
            return list(self._shards.items())

    def redistribute_keys(self) -> Dict[str, List[str]]: 1104102.h.kuaisou.com 
        """Return migration plan: source_node -> [keys to move]."""
        plan: Dict[str, List[str]] = {}
        for node_id, shard in self._shards.items(): 1104104.h.kuaisou.com 
            for key in shard.keys(): 1104105.h.kuaisou.com 
                target_node = self.hash_ring.get_node(key)
                if target_node and target_node != node_id: 1104109.h.kuaisou.com 
                    plan.setdefault(node_id, []).append(key)
        return plan

# distributed_cache/cache.py
import os
import json
import time
import threading
import logging
from typing import Any, Dict, List, Optional, Set, Callable
from distributed_cache.models import (
    CacheEntry, EvictionPolicy, LRUCache, LFUCache
)
from distributed_cache.sharding import ShardManager

logger = logging.getLogger(__name__)


class LocalCacheShard: 1104110.h.kuaisou.com 
    """Thread-safe local cache shard with eviction and expiry."""

    def __init__(
        self,
        max_size_bytes: int = 100 * 1024 * 1024,
        policy: EvictionPolicy = EvictionPolicy.LRU,
        cleanup_interval: float = 30.0,
    ):
        self.max_size_bytes = max_size_bytes
        self.policy = policy
        self.cleanup_interval = cleanup_interval

        if policy == EvictionPolicy.LFU: 1104112.h.kuaisou.com 
            self._store = LFUCache(max_size_bytes)
        else: 1104111.h.kuaisou.com 
            self._store = LRUCache(max_size_bytes)

        self._lock = threading.RLock(1104113.h.kuaisou.com)
        self._tags: Dict[str, Set[str]] = {}   # tag -> set of keys
        self._running = False
        self._cleanup_thread: Optional[threading.Thread] = None
        self._stats = {
            "hits": 0, "misses": 0, "puts": 0,
            "evictions": 0, "expirations": 0, "deletes": 0,
        }

    def start(self) -> None: 1104114.h.kuaisou.com 
        self._running = True
        self._cleanup_thread = threading.Thread(
            target=self._cleanup_loop, daemon=True, name="cache-cleanup"
        )
        self._cleanup_thread.start(1104115.h.kuaisou.com)

    def stop(self) -> None: 1104116.h.kuaisou.com 
        self._running = False
        if self._cleanup_thread: 1104117.h.kuaisou.com 
            self._cleanup_thread.join(timeout=5)

    def get(self, key: str) -> Optional[Any]: 1104118.h.kuaisou.com 
        with self._lock: 1104123.h.kuaisou.com 
            entry = self._store.get(key)
            if entry is None: 1104124.h.kuaisou.com 
                self._stats["misses"] += 1
                return None
            self._stats["hits"] += 1
            return entry.value

    def put(
        self, key: str, value: Any, ttl_seconds: float = 0,
        tags: Optional[List[str]] = None,
    ) -> Optional[CacheEntry]: 1107102.h.kuaisou.com 
        with self._lock: 1107104.h.kuaisou.com 
            entry = CacheEntry(
                key=key, value=value,
                ttl_seconds=ttl_seconds,
                tags=tags or [],
            )
            evicted = self._store.put(entry)
            self._stats["puts"] += 1
            if evicted: 1107105.h.kuaisou.com 
                self._stats["evictions"] += 1
                self._remove_tags(evicted.key)
            if entry.tags: 1107106.h.kuaisou.com 
                for tag in entry.tags: 1107125.h.kuaisou.com 
                    self._tags.setdefault(tag, set()).add(key)
            return evicted

    def delete(self, key: str) -> bool: 1107126.h.kuaisou.com 
        with self._lock: 1107127.h.kuaisou.com 
            entry = self._store.remove(key)
            if entry: 1107129.h.kuaisou.com 
                self._stats["deletes"] += 1
                self._remove_tags(key)
                return True
            return False

    def exists(self, key: str) -> bool: 1107130.h.kuaisou.com 
        with self._lock: 1107131.h.kuaisou.com 
            return self._store.get(key) is not None

    def ttl(self, key: str) -> float: 1107132.h.kuaisou.com 
        with self._lock: 1107134.h.kuaisou.com 
            entry = self._store.get(key)
            if entry is None: 1107135.h.kuaisou.com 
                return -1
            if entry.ttl_seconds <= 0: 1107136.h.kuaisou.com 
                return -1
            remaining = entry.expires_at - time.time(1107139.h.kuaisou.com)
            return max(0, remaining)

    def keys(self) -> List[str]: 1107143.h.kuaisou.com 
        with self._lock: 1107144.h.kuaisou.com 
            return self._store.keys(1107148.h.kuaisou.com)

    def size(self) -> int: 1107156.h.kuaisou.com 
        with self._lock: 1107155.h.kuaisou.com 
            return self._store.size(1107157.h.kuaisou.com)

    def current_bytes(self) -> int: 1107158.h.kuaisou.com 
        with self._lock:
            return self._store.current_bytes(1107159.h.kuaisou.com)

    def clear(self) -> None: 1107160.h.kuaisou.com 
        with self._lock: 1107163.h.kuaisou.com 
            self._store.clear(1107167.h.kuaisou.com)
            self._tags.clear(1107168.h.kuaisou.com)

    def invalidate_by_tag(self, tag: str) -> int: 1107170.h.kuaisou.com 
        with self._lock: 1107171.h.kuaisou.com 
            keys = self._tags.pop(tag, set()) 
            count = 0
            for key in keys: 1107172.h.kuaisou.comv 
                if self._store.remove(key): 1107173.h.kuaisou.com 
                    count += 1
            return count

    def get_stats(self) -> Dict[str, Any]: 1107174.h.kuaisou.com 
        with self._lock: 1107175.h.kuaisou.com 
            total = self._stats["hits"] + self._stats["misses"]
            hit_rate = self._stats["hits"] / total if total > 0 else 0
            return {
                **self._stats,
                "hit_rate": round(hit_rate, 4),
                "entries": self._store.size(1107176.h.kuaisou.com),
                "bytes_used": self._store.current_bytes(1107177.h.kuaisou.com),
                "max_bytes": self.max_size_bytes,
            }

    def _remove_tags(self, key: str) -> None: 1107178.h.kuaisou.com 
        for tag, key_set in list(self._tags.items()): 1107179.h.kuaisou.com 
            key_set.discard(key) 
            if not key_set: 1107180.h.kuaisou.com 
                del self._tags[tag]

    def _cleanup_loop(self) -> None: 1107181.h.kuaisou.com 
        while self._running: 1107182.h.kuaisou.com 
            time.sleep(self.cleanup_interval)
            self._cleanup_expired(1107183.h.kuaisou.com)

    def _cleanup_expired(self) -> None: 1107184.h.kuaisou.com 
        with self._lock: 1108106.h.kuaisou.com 
            now = time.time(1108107.h.kuaisou.com)
            expired_keys = []
            for key in self._store.keys(1108109.h.kuaisou.com): 1108108.h.kuaisou.com 
                entry = self._store.get(key)
                if entry and entry.is_expired(): 1108110.h.kuaisou.com 
                    expired_keys.append(key)
            for key in expired_keys: 1108112.h.kuaisou.com 
                self._store.remove(key)
                self._stats["expirations"] += 1
                self._remove_tags(key)
            if expired_keys: 1108113.h.kuaisou.com 
                logger.info("Cleaned up %d expired entries", len(expired_keys))


class MultiLevelCache: 1108114.h.kuaisou.com 
    """L1 (memory) + L2 (disk) + L3 (remote) cache."""

    def __init__(
        self,
        l1_max_bytes: int = 100 * 1024 * 1024,
        l2_dir: str = "/tmp/cache_l2",
        l2_max_bytes: int = 1 * 1024 * 1024 * 1024,
        remote_fetcher: Optional[Callable[[str], Optional[Any]]] = None,
    ):
        self.l1 = LocalCacheShard(max_size_bytes=l1_max_bytes)
        self.l2_dir = l2_dir
        self.l2_max_bytes = l2_max_bytes
        self.remote_fetcher = remote_fetcher
        self._lock = threading.RLock(1108115.h.kuaisou.com)
        self._pending: Dict[str, threading.Event] = {}
        os.makedirs(l2_dir, exist_ok=True)
        self.l1.start(1108116.h.kuaisou.com)

    def get(self, key: str) -> Optional[Any]: 1108117.h.kuaisou.com 
        # L1
        val = self.l1.get(key)
        if val is not None: 1108118.h.kuaisou.com 
            return val
        # L2
        val = self._l2_get(key)
        if val is not None: 1108119.h.kuaisou.com 
            self.l1.put(key, val)
            return val
        # L3 / remote
        if self.remote_fetcher: 1108120.h.kuaisou.com 
            # cache-aside with single-flight
            val = self._single_flight(key)
            if val is not None: 1108121.h.kuaisou.com 
                self.l1.put(key, val)
                self._l2_put(key, val)
            return val
        return None

    def put(
        self, key: str, value: Any, ttl_seconds: 1108122.h.kuaisou.com