Repository: LMCache/LMCache
File: lmcache/v1/storage_backend/abstract_backend.py (Python)

Code before:
# First Party
from lmcache.config import LMCacheEngineMetadata
from lmcache.utils import CacheEngineKey
from lmcache.v1.config import LMCacheEngineConfig
from lmcache.v1.lookup_server import LookupServerInterface
from lmcache.v1.memory_management import (
    MemoryAllocatorInterface,
    MemoryFormat,
    MemoryObj,
)
from lmcache.v1.storage_backend.storage_backend_listener import StorageBackendListener

if TYPE_CHECKING:
    # First Party
    from lmcache.v1.storage_backend import LocalCPUBackend


class StorageBackendInterface(metaclass=abc.ABCMeta):
    def __init__(
        self,
        dst_device: str = "cuda",
        config: Optional[LMCacheEngineConfig] = None,
        metadata: Optional[LMCacheEngineMetadata] = None,
        local_cpu_backend: Optional["LocalCPUBackend"] = None,
        lookup_server: Optional[LookupServerInterface] = None,
        loop: Optional[asyncio.AbstractEventLoop] = None,
    ):
        """
        Initialize the storage backend.

        :param dst_device: the device where the blocking retrieved KV is stored,
            could be either "cpu", "cuda", or "cuda:0", "cuda:1", etc.

        :raise: RuntimeError if the device is not valid
        """
        try:
            torch.device(dst_device)
        except RuntimeError:
            raise

        self.dst_device = dst_device
        self._listener: Optional[StorageBackendListener] = None

    def set_listener(self, listener: StorageBackendListener):
        """
        Set the listener to receive events.
        """
        self._listener = listener

    def _on_evict(self, keys: List[CacheEngineKey]) -> None:
        """

Review comment (performance):
Let's not include this for the default `StorageBackendInterface` since those parameters are pretty tightly coupled to each backend's implementation.

How about we create a base class for all the dynamic loading storage backend? Something like:
```python
class ExternalStorageBackend(StorageBackendInterface):  # I'm open to better class names.
    def __init__(
        self,
        dst_device: str = "cuda",
        config: Optional[LMCacheEngineConfig] = None,
        metadata: Optional[LMCacheEngineMetadata] = None,
        local_cpu_backend: Optional["LocalCPUBackend"] = None,
        lookup_server: Optional[LookupServerInterface] = None,
        loop: Optional[asyncio.AbstractEventLoop] = None,
    ):
        """
        All the external storage backends need to reimplement this function in order to
        override this function.
        """
```

Code after:
# First Party
from lmcache.config import LMCacheEngineMetadata
from lmcache.utils import CacheEngineKey
from lmcache.v1.config import LMCacheEngineConfig
from lmcache.v1.memory_management import (
    MemoryAllocatorInterface,
    MemoryFormat,
    MemoryObj,
)

if TYPE_CHECKING:
    # First Party
    from lmcache.v1.storage_backend import LocalCPUBackend


class StorageBackendInterface(metaclass=abc.ABCMeta):
    def __init__(
        self,
        dst_device: str = "cuda",
    ):
        """
        Initialize the storage backend.

        :param dst_device: the device where the blocking retrieved KV is stored,
            could be either "cpu", "cuda", or "cuda:0", "cuda:1", etc.

        :raise: RuntimeError if the device is not valid
        """
        try:
            torch.device(dst_device)
        except RuntimeError:
            raise

        self.dst_device = dst_device

    @abc.abstractmethod
    def contains(self, key: CacheEngineKey, pin: bool = False) -> bool:
        """
        Check whether key is in the storage backend.

        :param CacheEngineKey key: The key of the MemoryObj.

        :param bool pin: Whether to pin the key.
            If True, the corresponding KV cache will be
            pinned in the storage backend.

        :return: True if the key exists, False otherwise.
        """
        raise NotImplementedError

    @abc.abstractmethod