Caching Strategies and In Memory Storage Questions

import threading
import time
from functools import wraps
from typing import Any, Callable, Dict, Tuple

class _Entry:
    def __init__(self):
        self.value = None
        self.expiry = 0.0
        self.event = threading.Event()  # signals completion of computation

def ttl_cache(ttl_seconds: float):
    def decorator(fn: Callable):
        cache: Dict[Tuple[Any,...], _Entry] = {}
        cache_lock = threading.Lock()
        
        @wraps(fn)
        def wrapper(*args, **kwargs):
            key = (args, tuple(sorted(kwargs.items()))) if kwargs else (args, ())
            now = time.time()
            with cache_lock:
                entry = cache.get(key)
                if entry and entry.expiry > now:
                    return entry.value
                if entry and not entry.event.is_set():
                    # Another thread is computing; wait below after releasing lock
                    pass
                else:
                    # No entry or expired and not currently computing -> start computing
                    entry = _Entry()
                    entry.event.clear()
                    cache[key] = entry
                    # mark this thread as producer by leaving event unset
                    producer = True
                    # release lock and compute
                    # fallthrough
                    producer_flag = True
                    # we'll compute below outside lock
                    compute_entry = entry
                    break_marker = None
                    # use a sentinel to indicate producer path
                # If we got here, someone else is computing
            # Outside lock:
            # Wait for in-progress computation to finish and return result
            entry.event.wait()
            return entry.value

            # Producer path: (we intentionally structure so compute happens outside lock)
            # compute the value
            result = fn(*args, **kwargs)
            expiry = time.time() + ttl_seconds
            with cache_lock:
                compute_entry.value = result
                compute_entry.expiry = expiry
                compute_entry.event.set()
            return result

        # The above logic is slightly verbose to be explicit; simpler implementation below:
        def wrapper(*args, **kwargs):
            key = (args, tuple(sorted(kwargs.items()))) if kwargs else (args, ())
            while True:
                now = time.time()
                with cache_lock:
                    entry = cache.get(key)
                    if entry:
                        if entry.expiry > now and entry.event.is_set():
                            return entry.value
                        if not entry.event.is_set():
                            # someone is computing, take reference and wait
                            waiter = entry
                        else:
                            # expired and no active computation -> become producer
                            entry = _Entry()
                            cache[key] = entry
                            producer = True
                            break
                    else:
                        # absent -> become producer
                        entry = _Entry()
                        cache[key] = entry
                        producer = True
                        break
                # wait for in-progress computation
                waiter.event.wait()
                # loop to re-check expiry (in case entry was replaced)
            # producer does computation outside lock
            try:
                value = fn(*args, **kwargs)
                entry.value = value
                entry.expiry = time.time() + ttl_seconds
                return value
            finally:
                entry.event.set()
        return wrapper
    return decorator

import bisect
import hashlib

def _hash(value):
    return int(hashlib.md5(value.encode()).hexdigest(), 16)

class ConsistentHashRing:
    def __init__(self, replicas=100):
        self.replicas = replicas
        self.ring = []            # sorted list of hashes
        self.nodes = {}           # hash -> node

    def add_node(self, node):
        for i in range(self.replicas):
            h = _hash(f"{node}:{i}")
            if h in self.nodes: continue
            bisect.insort(self.ring, h)
            self.nodes[h] = node

    def remove_node(self, node):
        to_remove = []
        for i in range(self.replicas):
            h = _hash(f"{node}:{i}")
            if h in self.nodes and self.nodes[h] == node:
                to_remove.append(h)
        for h in to_remove:
            idx = bisect.bisect_left(self.ring, h)
            if idx < len(self.ring) and self.ring[idx] == h:
                self.ring.pop(idx)
            del self.nodes[h]

    def get_node(self, key):
        if not self.ring:
            return None
        h = _hash(str(key))
        idx = bisect.bisect_right(self.ring, h)
        if idx == len(self.ring):
            idx = 0
        return self.nodes[self.ring[idx]]

ring = ConsistentHashRing(replicas=200)
ring.add_node("hostA")
ring.add_node("hostB")
ring.add_node("hostC")

for key in ["user:1","user:42","session:xyz"]:
    print(key, "->", ring.get_node(key))