DSA to System Design Bridge

# Consistent Hashing Implementation
import hashlib
import bisect
from typing import Dict, List, Optional, Set

class ConsistentHashRing:
    def __init__(self, virtual_nodes: int = 150):
        self.ring: Dict[int, str] = {}
        self.servers: Set[str] = set()
        self.virtual_nodes = virtual_nodes
        self.sorted_hashes: List[int] = []
    
    def _hash(self, key: str) -> int:
        """MD5 hash function for consistent distribution"""
        return int(hashlib.md5(key.encode()).hexdigest()[:8], 16)
    
    def add_server(self, server: str) -> None:
        """Add server with virtual nodes for better distribution"""
        self.servers.add(server)
        
        for i in range(self.virtual_nodes):
            virtual_key = f"{server}:{i}"
            hash_val = self._hash(virtual_key)
            self.ring[hash_val] = server
            bisect.insort(self.sorted_hashes, hash_val)
        
        print(f"Added server {server} with {self.virtual_nodes} virtual nodes")
    
    def remove_server(self, server: str) -> None:
        """Remove server and all its virtual nodes"""
        self.servers.discard(server)
        
        # Remove all virtual nodes for this server
        to_remove = []
        for hash_val, srv in self.ring.items():
            if srv == server:
                to_remove.append(hash_val)
        
        for hash_val in to_remove:
            del self.ring[hash_val]
            self.sorted_hashes.remove(hash_val)
        
        print(f"Removed server {server}")
    
    def get_server(self, key: str) -> Optional[str]:
        """Find server responsible for given key"""
        if not self.ring:
            return None
        
        key_hash = self._hash(key)
        
        # Find first hash >= key_hash (clockwise on ring)
        idx = bisect.bisect_right(self.sorted_hashes, key_hash)
        if idx == len(self.sorted_hashes):
            idx = 0  # Wrap around
        
        return self.ring[self.sorted_hashes[idx]]
    
    def analyze_distribution(self, num_keys: int) -> Dict[str, int]:
        """Analyze how keys are distributed across servers"""
        distribution = {server: 0 for server in self.servers}
        
        for i in range(num_keys):
            key = f"key_{i}"
            server = self.get_server(key)
            if server:
                distribution[server] += 1
        
        return distribution

# Distributed Cache using Consistent Hashing
class DistributedCache:
    def __init__(self):
        self.hash_ring = ConsistentHashRing(200)
        self.connections = {}  # Mock connections
    
    def add_cache_server(self, server: str) -> None:
        self.hash_ring.add_server(server)
        self.connections[server] = {"host": server, "connected": True}
    
    def remove_cache_server(self, server: str) -> None:
        self.hash_ring.remove_server(server)
        self.connections.pop(server, None)
    
    def set(self, key: str, value) -> None:
        server = self.hash_ring.get_server(key)
        if not server:
            raise Exception("No servers available")
        
        # Mock cache set operation
        print(f"SET {key} → {value} on server {server}")
    
    def get(self, key: str):
        server = self.hash_ring.get_server(key)
        if not server:
            raise Exception("No servers available")
        
        # Mock cache get operation
        print(f"GET {key} from server {server}")
        return f"value_from_{server}"

# Consistent Hashing with Replication
class ReplicatedHashRing(ConsistentHashRing):
    def __init__(self, virtual_nodes: int = 150, replication_factor: int = 3):
        super().__init__(virtual_nodes)
        self.replication_factor = replication_factor
    
    def get_servers(self, key: str) -> List[str]:
        """Get multiple servers for replication"""
        if not self.ring or len(self.servers) == 0:
            return []
        
        key_hash = self._hash(key)
        servers = []
        seen_servers = set()
        
        # Start from first server >= key_hash
        idx = bisect.bisect_right(self.sorted_hashes, key_hash)
        
        while len(servers) < min(self.replication_factor, len(self.servers)):
            if idx >= len(self.sorted_hashes):
                idx = 0  # Wrap around
            
            server = self.ring[self.sorted_hashes[idx]]
            if server not in seen_servers:
                servers.append(server)
                seen_servers.add(server)
            
            idx += 1
        
        return servers

# Demo function
def demonstrate_consistent_hashing():
    print("=== Consistent Hashing Demo ===")
    
    # Basic consistent hashing
    ring = ConsistentHashRing(200)
    
    # Add servers
    servers = ["server-1", "server-2", "server-3"]
    for server in servers:
        ring.add_server(server)
    
    # Test key distribution
    print("\nKey → Server mapping:")
    test_keys = ["user:1001", "user:1002", "session:abc", "product:123"]
    for key in test_keys:
        server = ring.get_server(key)
        print(f"{key} → {server}")
    
    # Analyze distribution
    distribution = ring.analyze_distribution(10000)
    print(f"\nLoad distribution (10,000 keys):")
    for server, count in distribution.items():
        percentage = (count / 10000) * 100
        print(f"{server}: {count} keys ({percentage:.1f}%)")
    
    # Simulate server failure
    print(f"\n--- Removing {servers[1]} ---")
    ring.remove_server(servers[1])
    
    print("Key mapping after server removal:")
    for key in test_keys:
        server = ring.get_server(key)
        print(f"{key} → {server}")
    
    # Test replication
    print("\n=== Replication Demo ===")
    replicated_ring = ReplicatedHashRing(200, replication_factor=3)
    for server in ["node-1", "node-2", "node-3", "node-4"]:
        replicated_ring.add_server(server)
    
    for key in test_keys:
        servers = replicated_ring.get_servers(key)
        print(f"{key} → {servers}")