1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145

import hashlib
import bisect
from typing import Optional, List, Dict
 
class BasicConsistentHash:
    """
    Basic consistent hash ring implementation.
    
    Maps servers and keys to positions on a ring (0 to 2^32-1).
    Uses binary search for efficient O(log n) lookups.
    
    Limitation: Poor balance with few servers (solved by virtual nodes).
    """
    
    def __init__(self):
        # Sorted list of (position, server) pairs
        self._ring: List[int] = []
        self._position_to_server: Dict[int, str] = {}
    
    def _hash(self, key: str) -> int:
        """
        Hash a key to a position on the ring.
        
        Uses MD5 for good distribution. Returns value in [0, 2^32).
        """
        h = hashlib.md5(key.encode())
        # Take first 4 bytes as integer
        return int.from_bytes(h.digest()[:4], 'big')
    
    def add_server(self, server: str) -> None:
        """
        Add a server to the ring.
        
        Time Complexity: O(n) due to list insertion
        """
        position = self._hash(server)
        
        if position in self._position_to_server:
            # Collision - add suffix and rehash
            position = self._hash(f"{server}:collision")
        
        bisect.insort(self._ring, position)
        self._position_to_server[position] = server
    
    def remove_server(self, server: str) -> None:
        """
        Remove a server from the ring.
        
        Time Complexity: O(n) due to list removal
        """
        position = self._hash(server)
        
        if position in self._position_to_server:
            self._ring.remove(position)
            del self._position_to_server[position]
    
    def get_server(self, key: str) -> Optional[str]:
        """
        Get the server responsible for a key.
        
        Walks clockwise from key's position to find first server.
        
        Time Complexity: O(log n) using binary search
        """
        if not self._ring:
            return None
        
        position = self._hash(key)
        
        # Find first server position >= key position
        idx = bisect.bisect_left(self._ring, position)
        
        # Wrap around if past the end
        if idx >= len(self._ring):
            idx = 0
        
        server_position = self._ring[idx]
        return self._position_to_server[server_position]
    
    def get_servers_for_key(self, key: str, n: int = 3) -> List[str]:
        """
        Get n servers responsible for a key (for replication).
        
        Returns n distinct servers walking clockwise from key position.
        """
        if not self._ring:
            return []
        
        position = self._hash(key)
        idx = bisect.bisect_left(self._ring, position)
        
        servers = []
        seen = set()
        
        for i in range(len(self._ring)):
            server_pos = self._ring[(idx + i) % len(self._ring)]
            server = self._position_to_server[server_pos]
            
            if server not in seen:
                servers.append(server)
                seen.add(server)
                
                if len(servers) >= n:
                    break
        
        return servers
 
 
# Demonstration
if __name__ == "__main__":
    ch = BasicConsistentHash()
    
    # Add servers
    servers = ["server-a", "server-b", "server-c"]
    for s in servers:
        ch.add_server(s)
    
    # Generate some keys
    keys = [f"user-{i}" for i in range(10)]
    
    print("Initial distribution (3 servers):")
    print("-" * 50)
    distribution = {}
    for key in keys:
        server = ch.get_server(key)
        print(f"  {key} → {server}")
        distribution[server] = distribution.get(server, 0) + 1
    print(f"\nCounts: {distribution}")
    
    # Add a server
    print("\nAdding server-d:")
    ch.add_server("server-d")
    
    new_distribution = {}
    remapped = 0
    for key in keys:
        old_server = distribution.get(key)
        new_server = ch.get_server(key)
        new_distribution[new_server] = new_distribution.get(new_server, 0) + 1
        if old_server != new_server:
            remapped += 1
            print(f"  {key}: {old_server} → {new_server} (REMAPPED)")
    
    print(f"\nRemapped: {remapped}/{len(keys)} = {remapped/len(keys)*100:.0f}%")
    print(f"Expected: ~{100/4:.0f}% (1/n where n=4)")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205

import hashlib
import bisect
from typing import Optional, List, Dict, Set
from dataclasses import dataclass
import threading
 
@dataclass
class RingPosition:
    """Represents a position on the hash ring."""
    position: int
    physical_server: str
    virtual_node_id: int
 
class ConsistentHashRing:
    """
    Production-quality consistent hash ring with virtual nodes.
    
    Features:
    - Configurable virtual nodes per server
    - O(log n) lookups
    - Thread-safe operations
    - Server weight support via variable virtual nodes
    """
    
    def __init__(self, virtual_nodes: int = 150):
        """
        Initialize the hash ring.
        
        Args:
            virtual_nodes: Number of virtual nodes per physical server.
                          Higher = better balance, more memory.
                          Typical values: 100-300.
        """
        self._virtual_nodes = virtual_nodes
        self._ring: List[int] = []
        self._position_to_info: Dict[int, RingPosition] = {}
        self._servers: Set[str] = set()
        self._lock = threading.Lock()
    
    def _hash(self, key: str) -> int:
        """Hash key to 32-bit position."""
        h = hashlib.md5(key.encode())
        return int.from_bytes(h.digest()[:4], 'big')
    
    def _virtual_node_key(self, server: str, vnode_id: int) -> str:
        """Generate unique key for a virtual node."""
        return f"{server}#vnode{vnode_id}"
    
    def add_server(self, server: str, weight: int = 1) -> None:
        """
        Add a server with optional weight.
        
        Weight controls how many virtual nodes are created.
        weight=2 means twice the virtual nodes (twice the traffic).
        
        Time Complexity: O(v * log(n*v)) where v=virtual_nodes
        """
        with self._lock:
            if server in self._servers:
                return
            
            self._servers.add(server)
            
            # Create virtual nodes (scaled by weight)
            num_vnodes = self._virtual_nodes * weight
            
            for i in range(num_vnodes):
                vnode_key = self._virtual_node_key(server, i)
                position = self._hash(vnode_key)
                
                # Handle collisions by rehashing
                attempts = 0
                while position in self._position_to_info and attempts < 10:
                    position = self._hash(f"{vnode_key}:{attempts}")
                    attempts += 1
                
                if position not in self._position_to_info:
                    ring_pos = RingPosition(
                        position=position,
                        physical_server=server,
                        virtual_node_id=i
                    )
                    bisect.insort(self._ring, position)
                    self._position_to_info[position] = ring_pos
    
    def remove_server(self, server: str) -> None:
        """
        Remove a server and all its virtual nodes.
        
        Time Complexity: O(v * n) where v=virtual_nodes
        """
        with self._lock:
            if server not in self._servers:
                return
            
            self._servers.discard(server)
            
            # Remove all positions for this server
            positions_to_remove = [
                pos for pos, info in self._position_to_info.items()
                if info.physical_server == server
            ]
            
            for pos in positions_to_remove:
                self._ring.remove(pos)
                del self._position_to_info[pos]
    
    def get_server(self, key: str) -> Optional[str]:
        """
        Get the server responsible for a key.
        
        Time Complexity: O(log n)
        """
        with self._lock:
            if not self._ring:
                return None
            
            position = self._hash(key)
            idx = bisect.bisect_left(self._ring, position)
            
            if idx >= len(self._ring):
                idx = 0
            
            ring_pos = self._position_to_info[self._ring[idx]]
            return ring_pos.physical_server
    
    def get_servers(self, key: str, count: int = 3) -> List[str]:
        """
        Get multiple distinct servers for a key (for replication).
        """
        with self._lock:
            if not self._ring:
                return []
            
            position = self._hash(key)
            idx = bisect.bisect_left(self._ring, position)
            
            servers = []
            seen: Set[str] = set()
            
            for i in range(len(self._ring)):
                ring_idx = (idx + i) % len(self._ring)
                ring_pos = self._position_to_info[self._ring[ring_idx]]
                server = ring_pos.physical_server
                
                if server not in seen:
                    servers.append(server)
                    seen.add(server)
                    
                    if len(servers) >= count:
                        break
            
            return servers
    
    def get_distribution_stats(self) -> Dict[str, Dict]:
        """
        Get distribution statistics for monitoring.
        
        Returns per-server: virtual node count and ring coverage.
        """
        with self._lock:
            stats: Dict[str, Dict] = {}
            
            for server in self._servers:
                vnodes = sum(
                    1 for info in self._position_to_info.values()
                    if info.physical_server == server
                )
                stats[server] = {
                    "virtual_nodes": vnodes,
                    "coverage_pct": round(vnodes / len(self._ring) * 100, 2) if self._ring else 0
                }
            
            return stats
 
 
# Demonstration: Balance improvement with virtual nodes
if __name__ == "__main__":
    import random
    
    # Test with and without virtual nodes
    for vnodes in [1, 10, 100, 300]:
        ring = ConsistentHashRing(virtual_nodes=vnodes)
        
        servers = ["server-a", "server-b", "server-c", "server-d"]
        for s in servers:
            ring.add_server(s)
        
        # Distribute 10,000 random keys
        distribution = {s: 0 for s in servers}
        for i in range(10000):
            key = f"key-{random.randint(0, 1000000)}"
            server = ring.get_server(key)
            distribution[server] += 1
        
        # Calculate standard deviation
        values = list(distribution.values())
        mean = sum(values) / len(values)
        variance = sum((v - mean) ** 2 for v in values) / len(values)
        std_dev = variance ** 0.5
        
        print(f"\nVirtual nodes: {vnodes}")
        print(f"  Distribution: {distribution}")
        print(f"  Std deviation: {std_dev:.0f} (lower is better)")
        print(f"  Balance score: {(mean - std_dev) / mean * 100:.1f}%")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66

# NGINX Consistent Hashing Configuration
 
upstream backend_consistent_hash {
    # Enable consistent hashing based on request URI
    hash $request_uri consistent;
    
    # Server list - order doesn't matter with consistent hash
    server 10.0.1.1:8080;
    server 10.0.1.2:8080;
    server 10.0.1.3:8080;
    
    # Weight affects virtual node count
    # Higher weight = more virtual nodes = more traffic
    # server 10.0.1.4:8080 weight=2;
}
 
upstream consistent_hash_by_ip {
    # Hash by client IP for session affinity
    hash $remote_addr consistent;
    
    server 10.0.1.1:8080;
    server 10.0.1.2:8080;
    server 10.0.1.3:8080;
}
 
upstream consistent_hash_custom {
    # Hash by custom variable
    # Cookie, header, or computed value
    hash $cookie_session_id consistent;
    
    server 10.0.1.1:8080;
    server 10.0.1.2:8080;
    server 10.0.1.3:8080;
}
 
# Consistent hash with health checking
# When server goes down, only its keys remap
upstream consistent_with_backup {
    hash $request_uri consistent;
    
    server 10.0.1.1:8080;
    server 10.0.1.2:8080;
    server 10.0.1.3:8080;
    
    # down marker: keys remap to other servers
    # server 10.0.1.4:8080 down;
    
    # backup: only receives traffic if primary server fails
    # server 10.0.1.5:8080 backup;
}
 
server {
    listen 80;
    
    location /api/ {
        proxy_pass http://backend_consistent_hash;
        
        # Retry on failure - request may remap to different server
        proxy_next_upstream error timeout http_502 http_503;
        proxy_next_upstream_tries 2;
    }
    
    location /session/ {
        proxy_pass http://consistent_hash_by_ip;
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59

# HAProxy Consistent Hashing Configuration
 
global
    maxconn 50000
 
defaults
    mode http
    timeout connect 5s
    timeout client 30s
    timeout server 60s
 
backend consistent_hash_url
    # Consistent hash on URL
    balance uri
    hash-type consistent
    
    server app1 10.0.1.1:8080 check
    server app2 10.0.1.2:8080 check
    server app3 10.0.1.3:8080 check
 
backend consistent_hash_source
    # Consistent hash on source IP
    balance source
    hash-type consistent
    
    server app1 10.0.1.1:8080 check
    server app2 10.0.1.2:8080 check
    server app3 10.0.1.3:8080 check
 
backend consistent_hash_header
    # Consistent hash on custom header
    balance hdr(X-User-ID)
    hash-type consistent
    
    # Weight affects traffic share
    server app1 10.0.1.1:8080 weight 100 check
    server app2 10.0.1.2:8080 weight 100 check
    server app3 10.0.1.3:8080 weight 50 check  # Half traffic
 
backend consistent_hash_cookie
    # Hash on session cookie
    balance url_param sessionid
    hash-type consistent
    
    server app1 10.0.1.1:8080 check
    server app2 10.0.1.2:8080 check
 
frontend http_front
    bind *:80
    
    # Route based on path
    acl is_api path_beg /api
    use_backend consistent_hash_url if is_api
    default_backend consistent_hash_source
 
listen stats
    bind *:8404
    stats enable
    stats uri /stats

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78

# Envoy Consistent Hashing Configuration
 
static_resources:
  clusters:
  - name: backend_service
    type: STRICT_DNS
    connect_timeout: 5s
    
    # Ring hash load balancing (consistent hashing)
    lb_policy: RING_HASH
    
    ring_hash_lb_config:
      # Minimum ring size (virtual nodes)
      minimum_ring_size: 1024
      # Maximum ring size
      maximum_ring_size: 8388608
      # Hash function
      hash_function: XX_HASH  # or MURMUR_HASH_2
    
    load_assignment:
      cluster_name: backend_service
      endpoints:
      - lb_endpoints:
        - endpoint:
            address:
              socket_address:
                address: 10.0.1.1
                port_value: 8080
        - endpoint:
            address:
              socket_address:
                address: 10.0.1.2
                port_value: 8080
        - endpoint:
            address:
              socket_address:
                address: 10.0.1.3
                port_value: 8080
    
    health_checks:
    - timeout: 3s
      interval: 5s
      unhealthy_threshold: 3
      healthy_threshold: 2
      http_health_check:
        path: /health
 
  listeners:
  - name: listener_0
    address:
      socket_address:
        address: 0.0.0.0
        port_value: 80
    filter_chains:
    - filters:
      - name: envoy.filters.network.http_connection_manager
        typed_config:
          "@type": type.googleapis.com/envoy.extensions.filters.network.http_connection_manager.v3.HttpConnectionManager
          stat_prefix: ingress_http
          route_config:
            name: local_route
            virtual_hosts:
            - name: local_service
              domains: ["*"]
              routes:
              - match:
                  prefix: "/"
                route:
                  cluster: backend_service
                  hash_policy:
                  # Hash on header
                  - header:
                      header_name: x-user-id
                  # Fallback: hash on source IP
                  - connection_properties:
                      source_ip: true
          http_filters:
          - name: envoy.filters.http.router

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70

def jump_hash(key: int, num_buckets: int) -> int:
    """
    Google's Jump Hash algorithm.
    
    Fast, memory-efficient consistent hashing.
    Returns bucket number in [0, num_buckets).
    
    Properties:
    - O(log n) time
    - O(1) space
    - Perfectly uniform distribution
    - Minimal remapping on bucket count changes
    
    Limitation: Only works for numbered buckets 0..n-1
    Removing bucket k requires renumbering k+1..n
    
    Reference: https://arxiv.org/abs/1406.2294
    """
    if num_buckets <= 0:
        raise ValueError("num_buckets must be positive")
    
    b = -1
    j = 0
    
    while j < num_buckets:
        b = j
        key = (key * 2862933555777941757 + 1) & 0xFFFFFFFFFFFFFFFF
        j = int((b + 1) * (float(1 << 31) / float((key >> 33) + 1)))
    
    return b
 
 
def string_to_int(s: str) -> int:
    """Convert string to int for hashing."""
    import hashlib
    h = hashlib.md5(s.encode())
    return int.from_bytes(h.digest()[:8], 'big')
 
 
# Demonstration
if __name__ == "__main__":
    print("Jump Hash Distribution Test")
    print("=" * 50)
    
    # Test distribution with 5 buckets
    buckets = 5
    counts = [0] * buckets
    
    for i in range(10000):
        key = string_to_int(f"key-{i}")
        bucket = jump_hash(key, buckets)
        counts[bucket] += 1
    
    print(f"\n5 buckets, 10,000 keys:")
    for i, count in enumerate(counts):
        bar = "█" * (count // 100)
        print(f"  Bucket {i}: {count:5d} {bar}")
    
    # Test remapping when adding bucket
    print(f"\nRemapping test (5 → 6 buckets):")
    remapped = 0
    for i in range(10000):
        key = string_to_int(f"key-{i}")
        old = jump_hash(key, 5)
        new = jump_hash(key, 6)
        if old != new:
            remapped += 1
    
    print(f"  Remapped: {remapped}/10,000 = {remapped/100:.1f}%")
    print(f"  Expected: ~{100/6:.1f}% (1/n)")