What Are Data Structures? — Senior Level¶

import hashlib
import math
import queue
import threading
import time
from collections import defaultdict
from typing import Any


# --- Thread-safe queue for producer-consumer ---

def producer_consumer_example():
    task_queue = queue.Queue(maxsize=100)
    results = []
    results_lock = threading.Lock()

    def producer(n_tasks: int):
        for i in range(n_tasks):
            task_queue.put({"id": i, "data": f"payload-{i}"})
        task_queue.put(None)  # sentinel to stop consumer

    def consumer():
        while True:
            task = task_queue.get()  # blocks if empty
            if task is None:
                task_queue.task_done()
                break
            # Simulate processing
            result = task["id"] ** 2
            with results_lock:
                results.append(result)
            task_queue.task_done()

    t1 = threading.Thread(target=producer, args=(50,))
    t2 = threading.Thread(target=consumer)
    t1.start()
    t2.start()
    t1.join()
    t2.join()

    print(f"Processed {len(results)} tasks, sum={sum(results)}")


# --- Thread-safe Bloom Filter ---

class ThreadSafeBloomFilter:
    """Bloom filter with thread-safe add and check operations."""

    def __init__(self, expected_items: int, false_positive_rate: float = 0.01):
        self._size = self._optimal_size(expected_items, false_positive_rate)
        self._hash_count = self._optimal_hashes(self._size, expected_items)
        self._bit_array = bytearray(self._size)
        self._lock = threading.Lock()
        self._count = 0

    @staticmethod
    def _optimal_size(n: int, p: float) -> int:
        return int(-n * math.log(p) / (math.log(2) ** 2))

    @staticmethod
    def _optimal_hashes(m: int, n: int) -> int:
        return max(1, int((m / n) * math.log(2)))

    def _hashes(self, item: str) -> list[int]:
        positions = []
        for i in range(self._hash_count):
            digest = hashlib.sha256(f"{item}:{i}".encode()).hexdigest()
            positions.append(int(digest, 16) % self._size)
        return positions

    def add(self, item: str) -> None:
        positions = self._hashes(item)
        with self._lock:
            for pos in positions:
                self._bit_array[pos] = 1
            self._count += 1

    def might_contain(self, item: str) -> bool:
        positions = self._hashes(item)
        with self._lock:
            return all(self._bit_array[pos] for pos in positions)

    @property
    def count(self) -> int:
        with self._lock:
            return self._count


# --- Thread-safe LRU Cache ---

class ThreadSafeLRUCache:
    """LRU cache with thread safety for concurrent access."""

    def __init__(self, capacity: int):
        self._capacity = capacity
        self._cache: dict[str, Any] = {}
        self._order: list[str] = []  # least recent first
        self._lock = threading.Lock()
        self._hits = 0
        self._misses = 0

    def get(self, key: str) -> Any | None:
        with self._lock:
            if key in self._cache:
                self._hits += 1
                self._order.remove(key)
                self._order.append(key)
                return self._cache[key]
            self._misses += 1
            return None

    def put(self, key: str, value: Any) -> None:
        with self._lock:
            if key in self._cache:
                self._order.remove(key)
            elif len(self._cache) >= self._capacity:
                evicted = self._order.pop(0)
                del self._cache[evicted]
            self._cache[key] = value
            self._order.append(key)

    @property
    def hit_rate(self) -> float:
        total = self._hits + self._misses
        return self._hits / total if total > 0 else 0.0

    @property
    def stats(self) -> dict:
        with self._lock:
            return {
                "size": len(self._cache),
                "capacity": self._capacity,
                "hits": self._hits,
                "misses": self._misses,
                "hit_rate": f"{self.hit_rate:.2%}",
            }


if __name__ == "__main__":
    producer_consumer_example()

    # Bloom filter demo
    bf = ThreadSafeBloomFilter(expected_items=10000, false_positive_rate=0.01)
    for i in range(1000):
        bf.add(f"user:{i}")
    print(f"Contains 'user:42': {bf.might_contain('user:42')}")   # True
    print(f"Contains 'user:9999': {bf.might_contain('user:9999')}")  # Likely False
    print(f"Items added: {bf.count}")

    # LRU cache demo
    cache = ThreadSafeLRUCache(capacity=100)
    for i in range(200):
        cache.put(f"key:{i}", f"value:{i}")
    for i in range(150, 200):
        cache.get(f"key:{i}")
    print(f"Cache stats: {cache.stats}")

from prometheus_client import Counter, Gauge, Histogram, start_http_server

cache_hits = Counter("cache_hits_total", "Total cache hits", ["cache_name"])
cache_misses = Counter("cache_misses_total", "Total cache misses", ["cache_name"])
queue_depth = Gauge("queue_depth", "Current items in queue", ["queue_name"])
op_duration = Histogram(
    "datastructure_operation_seconds",
    "Latency of data structure operations",
    ["structure", "operation"],
    buckets=(0.0001, 0.0005, 0.001, 0.005, 0.01, 0.05, 0.1, 0.5, 1.0),
)


# Usage:
# cache_hits.labels(cache_name="user-cache").inc()
# queue_depth.labels(queue_name="task-queue").set(current_size)
# with op_duration.labels(structure="bloom_filter", operation="check").time():
#     result = bloom_filter.might_contain(key)


if __name__ == "__main__":
    start_http_server(9090)
    print("Metrics server running on :9090/metrics")

import threading
import time
from typing import Any, Callable


class SingleFlight:
    """Ensures only one thread fetches a given key at a time."""

    def __init__(self):
        self._lock = threading.Lock()
        self._in_flight: dict[str, threading.Event] = {}
        self._results: dict[str, Any] = {}

    def do(self, key: str, fn: Callable[[], Any]) -> Any:
        with self._lock:
            if key in self._in_flight:
                event = self._in_flight[key]
                # Release lock, wait for the in-flight request
                self._lock.release()
                event.wait()
                self._lock.acquire()
                return self._results.pop(key, None)

            event = threading.Event()
            self._in_flight[key] = event

        try:
            result = fn()
            with self._lock:
                self._results[key] = result
        finally:
            event.set()
            with self._lock:
                del self._in_flight[key]

        return result


if __name__ == "__main__":
    sf = SingleFlight()
    fetch_count = 0
    count_lock = threading.Lock()

    def fetch_from_db() -> str:
        global fetch_count
        with count_lock:
            fetch_count += 1
        print("Actually fetching from DB (should print once)")
        time.sleep(0.1)
        return "db-result"

    threads = []
    for _ in range(100):
        t = threading.Thread(target=sf.do, args=("popular-key", fetch_from_db))
        threads.append(t)
        t.start()

    for t in threads:
        t.join()

    print(f"Total DB fetches: {fetch_count}")  # Should be 1