pydvl.utils.caching.memcached

MemcachedClientConfig dataclass

MemcachedClientConfig(
    server: Tuple[str, int] = ("localhost", 11211),
    connect_timeout: float = 1.0,
    timeout: float = 1.0,
    no_delay: bool = True,
    serde: PickleSerde = PickleSerde(pickle_version=PICKLE_VERSION),
)

Configuration of the memcached client.

PARAMETER	DESCRIPTION
server	A tuple of (IP|domain name, port). TYPE: Tuple[str, int] DEFAULT: ('localhost', 11211)
connect_timeout	How many seconds to wait before raising ConnectionRefusedError on failure to connect. TYPE: float DEFAULT: 1.0
timeout	Duration in seconds to wait for send or recv calls on the socket connected to memcached. TYPE: float DEFAULT: 1.0
no_delay	If True, set the TCP_NODELAY flag, which may help with performance in some cases. TYPE: bool DEFAULT: True
serde	Serializer / Deserializer ("serde"). The default PickleSerde should work in most cases. See pymemcache.client.base.Client for details. TYPE: PickleSerde DEFAULT: PickleSerde(pickle_version=PICKLE_VERSION)

MemcachedCacheBackend

MemcachedCacheBackend(config: MemcachedClientConfig = MemcachedClientConfig())

Bases: CacheBackend

Memcached cache backend for the distributed caching of functions.

Implements the CacheBackend interface for a memcached based cache. This allows sharing evaluations across processes and nodes in a cluster. You can run memcached as a service, locally or remotely, see the caching documentation.

PARAMETER	DESCRIPTION
config	Memcached client configuration. TYPE: MemcachedClientConfig DEFAULT: MemcachedClientConfig()

ATTRIBUTE	DESCRIPTION
config	Memcached client configuration.
client	Memcached client instance.

Example

Basic usage:

>>> from pydvl.utils.caching.memcached import MemcachedCacheBackend
>>> cache_backend = MemcachedCacheBackend()
>>> cache_backend.clear()
>>> value = 42
>>> cache_backend.set("key", value)
>>> cache_backend.get("key")
42

Callable wrapping:

>>> from pydvl.utils.caching.memcached import MemcachedCacheBackend
>>> cache_backend = MemcachedCacheBackend()
>>> cache_backend.clear()
>>> value = 42
>>> def foo(x: int):
...     return x + 1
...
>>> wrapped_foo = cache_backend.wrap(foo)
>>> wrapped_foo(value)
43
>>> wrapped_foo.stats.misses
1
>>> wrapped_foo.stats.hits
0
>>> wrapped_foo(value)
43
>>> wrapped_foo.stats.misses
1
>>> wrapped_foo.stats.hits
1

PARAMETER	DESCRIPTION
config	Memcached client configuration. TYPE: MemcachedClientConfig DEFAULT: MemcachedClientConfig()

Source code in src/pydvl/utils/caching/memcached.py

def __init__(self, config: MemcachedClientConfig = MemcachedClientConfig()) -> None:
    """Initialize memcached cache backend.

    Args:
        config: Memcached client configuration.
    """

    super().__init__()
    self.config = config
    self.client = self._connect(self.config)

wrap

wrap(
    func: Callable, *, config: Optional[CachedFuncConfig] = None
) -> CachedFunc

Wraps a function to cache its results.

PARAMETER	DESCRIPTION
func	The function to wrap. TYPE: Callable
config	Optional caching options for the wrapped function. TYPE: Optional[CachedFuncConfig] DEFAULT: None

RETURNS	DESCRIPTION
CachedFunc	The wrapped cached function.

Source code in src/pydvl/utils/caching/base.py

def wrap(
    self,
    func: Callable,
    *,
    config: Optional[CachedFuncConfig] = None,
) -> "CachedFunc":
    """Wraps a function to cache its results.

    Args:
        func: The function to wrap.
        config: Optional caching options for the wrapped function.

    Returns:
        The wrapped cached function.
    """
    return CachedFunc(
        func,
        cache_backend=self,
        config=config,
    )

get

get(key: str) -> Optional[Any]

Get value from memcached.

PARAMETER	DESCRIPTION
key	Cache key. TYPE: str

RETURNS	DESCRIPTION
Optional[Any]	Cached value or None if not found or client disconnected.

Source code in src/pydvl/utils/caching/memcached.py

def get(self, key: str) -> Optional[Any]:
    """Get value from memcached.

    Args:
        key: Cache key.

    Returns:
        Cached value or None if not found or client disconnected.
    """
    result = None
    try:
        result = self.client.get(key)
    except socket.timeout as e:
        self.stats.timeouts += 1
        warnings.warn(f"{type(self).__name__}: {str(e)}", RuntimeWarning)
    except OSError as e:
        self.stats.errors += 1
        warnings.warn(f"{type(self).__name__}: {str(e)}", RuntimeWarning)
    except AttributeError as e:
        # FIXME: this depends on _recv() failing on invalid sockets
        # See pymemcache.base.py,
        self.stats.reconnects += 1
        warnings.warn(f"{type(self).__name__}: {str(e)}", RuntimeWarning)
        self.client = self._connect(self.config)
    if result is None:
        self.stats.misses += 1
    else:
        self.stats.hits += 1
    return result

set

set(key: str, value: Any) -> None

Set value in memcached.

PARAMETER	DESCRIPTION
key	Cache key. TYPE: str
value	Value to cache. TYPE: Any

Source code in src/pydvl/utils/caching/memcached.py

def set(self, key: str, value: Any) -> None:
    """Set value in memcached.

    Args:
        key: Cache key.
        value: Value to cache.
    """
    self.client.set(key, value, noreply=True)
    self.stats.sets += 1

clear

clear() -> None

Flush all values from memcached.

Source code in src/pydvl/utils/caching/memcached.py

def clear(self) -> None:
    """Flush all values from memcached."""
    self.client.flush_all(noreply=True)

combine_hashes

combine_hashes(*args: str) -> str

Join cache key components for Memcached.

Source code in src/pydvl/utils/caching/memcached.py

def combine_hashes(self, *args: str) -> str:
    """Join cache key components for Memcached."""
    return ":".join(args)

__getstate__

__getstate__() -> Dict

Enables pickling after a socket has been opened to the memcached server, by removing the client from the stored data.

Source code in src/pydvl/utils/caching/memcached.py

def __getstate__(self) -> Dict:
    """Enables pickling after a socket has been opened to the
    memcached server, by removing the client from the stored
    data."""
    odict = self.__dict__.copy()
    del odict["client"]
    return odict

__setstate__

__setstate__(d: Dict)

Restores a client connection after loading from a pickle.

Source code in src/pydvl/utils/caching/memcached.py

def __setstate__(self, d: Dict):
    """Restores a client connection after loading from a pickle."""
    self.config = d["config"]
    self.stats = d["stats"]
    self.client = self._connect(self.config)