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

import threading
from dataclasses import dataclass
from typing import Optional
import heapq
 
@dataclass
class ServerState:
    """Tracks connection state for a single server."""
    address: str
    active_connections: int = 0
    healthy: bool = True
 
class LeastConnectionsLoadBalancer:
    """
    Least Connections load balancer implementation.
    
    Uses a simple linear scan to find the server with fewest
    connections. For small server counts (< 100), this is efficient.
    """
    
    def __init__(self, servers: list[str]):
        """
        Initialize with list of backend server addresses.
        
        Args:
            servers: List of server identifiers
        """
        self._servers = {
            addr: ServerState(address=addr) 
            for addr in servers
        }
        self._lock = threading.Lock()
    
    def get_next_server(self) -> Optional[str]:
        """
        Select server with fewest active connections.
        
        Time Complexity: O(n) where n is number of servers
        
        Returns:
            Server address with minimum connections, or None if all unhealthy
        """
        with self._lock:
            # Filter to healthy servers only
            healthy_servers = [
                s for s in self._servers.values() 
                if s.healthy
            ]
            
            if not healthy_servers:
                return None
            
            # Find server with minimum connections
            # Use address as tiebreaker for determinism
            best = min(
                healthy_servers,
                key=lambda s: (s.active_connections, s.address)
            )
            
            # Increment connection count
            best.active_connections += 1
            
            return best.address
    
    def release_connection(self, address: str) -> None:
        """
        Called when a connection completes.
        
        CRITICAL: This must be called for every connection,
        or the load balancer state becomes corrupted!
        """
        with self._lock:
            if address in self._servers:
                server = self._servers[address]
                server.active_connections = max(0, server.active_connections - 1)
    
    def mark_unhealthy(self, address: str) -> None:
        """Mark a server as unhealthy."""
        with self._lock:
            if address in self._servers:
                self._servers[address].healthy = False
                # Reset connections - existing requests will still complete
                self._servers[address].active_connections = 0
    
    def mark_healthy(self, address: str) -> None:
        """Mark a server as healthy."""
        with self._lock:
            if address in self._servers:
                self._servers[address].healthy = True
    
    def get_connection_counts(self) -> dict[str, int]:
        """Get current connection counts for monitoring."""
        with self._lock:
            return {
                addr: s.active_connections 
                for addr, s in self._servers.items()
            }
 
 
# Context manager for clean connection lifecycle
class ConnectionContext:
    """
    Ensures connection is properly released after use.
    
    Usage:
        with ConnectionContext(lb) as server:
            # Make request to server
            response = make_request(server)
    """
    
    def __init__(self, lb: LeastConnectionsLoadBalancer):
        self._lb = lb
        self._server = None
    
    def __enter__(self) -> Optional[str]:
        self._server = self._lb.get_next_server()
        return self._server
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        if self._server:
            self._lb.release_connection(self._server)
        return False  # Don't suppress exceptions
 
 
# Example usage
if __name__ == "__main__":
    import time
    import random
    
    servers = ["10.0.1.1:8080", "10.0.1.2:8080", "10.0.1.3:8080"]
    lb = LeastConnectionsLoadBalancer(servers)
    
    # Simulate variable workload
    print("Simulating variable request durations:")
    print("-" * 50)
    
    for i in range(10):
        with ConnectionContext(lb) as server:
            # Simulate variable processing time
            duration = random.uniform(0.1, 1.0)
            print(f"Request {i+1}: {server} "
                  f"(counts: {lb.get_connection_counts()})")
            time.sleep(duration)
    
    print("-" * 50)
    print(f"Final state: {lb.get_connection_counts()}")

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

import threading
from dataclasses import dataclass
from typing import Optional
 
@dataclass
class WeightedServer:
    """Server with weight and connection tracking."""
    address: str
    weight: int
    active_connections: int = 0
    healthy: bool = True
    
    @property
    def load_ratio(self) -> float:
        """
        Calculate load relative to capacity.
        
        Lower ratio = server has more available capacity.
        Weight of 0 returns infinity to avoid selection.
        """
        if self.weight == 0:
            return float('inf')
        return self.active_connections / self.weight
 
class WeightedLeastConnectionsLB:
    """
    Weighted Least Connections load balancer.
    
    Selects the server with the lowest connections/weight ratio,
    effectively choosing the server with most available capacity.
    """
    
    def __init__(self, servers: dict[str, int]):
        """
        Initialize with servers and their weights.
        
        Args:
            servers: Dict mapping server address to weight
        """
        self._servers = {
            addr: WeightedServer(address=addr, weight=w)
            for addr, w in servers.items()
        }
        self._lock = threading.Lock()
    
    def get_next_server(self) -> Optional[str]:
        """
        Select server with lowest load ratio.
        
        Time Complexity: O(n)
        
        Returns:
            Server address, or None if no healthy servers
        """
        with self._lock:
            healthy_servers = [
                s for s in self._servers.values()
                if s.healthy and s.weight > 0
            ]
            
            if not healthy_servers:
                return None
            
            # Select by minimum load ratio
            # Use address as tiebreaker for determinism
            best = min(
                healthy_servers,
                key=lambda s: (s.load_ratio, s.address)
            )
            
            best.active_connections += 1
            return best.address
    
    def release_connection(self, address: str) -> None:
        """Release a connection when request completes."""
        with self._lock:
            if address in self._servers:
                server = self._servers[address]
                server.active_connections = max(0, server.active_connections - 1)
    
    def update_weight(self, address: str, new_weight: int) -> None:
        """
        Update server weight dynamically.
        
        Useful for:
        - Responding to detected capacity changes
        - Gradual scaling up/down
        - Manual traffic shifts
        """
        with self._lock:
            if address in self._servers:
                self._servers[address].weight = max(0, new_weight)
    
    def get_states(self) -> list[dict]:
        """Get current state for monitoring/debugging."""
        with self._lock:
            return [
                {
                    "address": s.address,
                    "weight": s.weight,
                    "connections": s.active_connections,
                    "load_ratio": round(s.load_ratio, 3),
                    "healthy": s.healthy,
                }
                for s in self._servers.values()
            ]
 
 
# Demonstration
if __name__ == "__main__":
    # Heterogeneous cluster: different capacity servers
    servers = {
        "large-instance": 10,   # High capacity
        "medium-instance": 5,   # Medium capacity  
        "small-instance": 2,    # Lower capacity
    }
    
    lb = WeightedLeastConnectionsLB(servers)
    
    print("Initial state:")
    for state in lb.get_states():
        print(f"  {state}")
    
    print("\nDistributing 17 requests (proportional to weights 10:5:2):")
    
    selections = []
    for i in range(17):
        server = lb.get_next_server()
        selections.append(server)
    
    # Count selections
    from collections import Counter
    counts = Counter(selections)
    print(f"\nDistribution: {dict(counts)}")
    print("Expected ratio: 10:5:2")
    
    # Show final state (all connections still active)
    print("\nFinal state (connections active):")
    for state in lb.get_states():
        print(f"  {state}")

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

# NGINX Least Connections Configuration
 
upstream backend_pool {
    # Enable least_conn algorithm
    least_conn;
    
    # Server list - NGINX tracks connections to each
    server 10.0.1.1:8080;
    server 10.0.1.2:8080;
    server 10.0.1.3:8080;
    
    # With health checking (NGINX Plus)
    # health_check interval=5s fails=3 passes=2;
}
 
upstream weighted_least_conn {
    # Combine least_conn with weights
    # NGINX uses weighted least connections when weights differ
    least_conn;
    
    # High-capacity servers
    server 10.0.1.1:8080 weight=10;
    server 10.0.1.2:8080 weight=10;
    
    # Medium capacity
    server 10.0.2.1:8080 weight=5;
    
    # Lower capacity
    server 10.0.3.1:8080 weight=2;
    
    # Connection limits per server (optional)
    # Prevents any single server from being overwhelmed
    # server 10.0.1.1:8080 weight=10 max_conns=1000;
}
 
upstream with_keepalive {
    least_conn;
    
    server 10.0.1.1:8080;
    server 10.0.1.2:8080;
    
    # Keep idle connections to backends
    # Reduces connection overhead for HTTP/1.1
    keepalive 32;
    
    # For HTTP/2 backends
    # keepalive_requests 1000;
    # keepalive_timeout 60s;
}
 
server {
    listen 80;
    
    location / {
        proxy_pass http://backend_pool;
        
        # Required for keepalive to work
        proxy_http_version 1.1;
        proxy_set_header Connection "";
        
        # Timeouts affect connection counts
        # Long timeouts = connections held longer
        proxy_connect_timeout 5s;
        proxy_read_timeout 60s;
        proxy_send_timeout 60s;
    }
    
    # Status page showing connection distribution (NGINX Plus)
    location /api/status {
        # api write=off;
        # allow 127.0.0.1;
        # deny all;
    }
}

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

# HAProxy Least Connections Configuration
 
global
    maxconn 50000
 
defaults
    mode http
    timeout connect 5s
    timeout client 30s
    timeout server 30s
    option httplog
 
backend app_servers_leastconn
    # Least connections balancing
    balance leastconn
    
    server app1 10.0.1.1:8080 check maxconn 2000
    server app2 10.0.1.2:8080 check maxconn 2000
    server app3 10.0.1.3:8080 check maxconn 2000
    
    # maxconn: Maximum connections per server
    # Queues requests when limit reached
 
backend weighted_leastconn
    # Weighted least connections
    balance leastconn
    
    # Weight affects the connection/weight ratio calculation
    server large1 10.0.1.1:8080 weight 100 check
    server large2 10.0.1.2:8080 weight 100 check
    server medium1 10.0.2.1:8080 weight 50 check
    server small1 10.0.3.1:8080 weight 25 check
 
backend with_queue_management
    balance leastconn
    
    # fullconn: Expected max connections for queue calculation
    fullconn 5000
    
    server app1 10.0.1.1:8080 weight 100 check maxconn 1000
    server app2 10.0.1.2:8080 weight 100 check maxconn 1000
    
    # Queue management
    # When all servers at maxconn, requests queue here
    timeout queue 30s
    
    # Retries on connection failure
    retries 3
    option redispatch  # Allow retry on different server
 
frontend http_front
    bind *:80
    default_backend app_servers_leastconn
 
# Very useful: stats page shows live connection counts
listen stats
    bind *:8404
    stats enable
    stats uri /stats
    stats refresh 5s
    
    # Shows per-server:
    # - Current connections
    # - Connection rate
    # - Bytes in/out
    # - Response times

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

import time
from dataclasses import dataclass, field
from typing import Optional
import threading
 
@dataclass
class ServerWithSlowStart:
    """Server with slow-start support to prevent thundering herd."""
    address: str
    weight: int = 1
    active_connections: int = 0
    healthy: bool = True
    
    # Slow start tracking
    _startup_time: Optional[float] = None
    _slow_start_duration: float = 60.0  # seconds
    
    def mark_healthy(self):
        """Called when server becomes healthy."""
        self.healthy = True
        self._startup_time = time.time()
    
    @property
    def effective_weight(self) -> float:
        """
        Weight adjusted for slow-start period.
        
        Gradually increases from 10% to 100% over slow_start_duration.
        """
        if not self._startup_time:
            return float(self.weight)
        
        elapsed = time.time() - self._startup_time
        
        if elapsed >= self._slow_start_duration:
            # Slow start complete
            self._startup_time = None
            return float(self.weight)
        
        # Linear ramp from 0.1 to 1.0
        progress = elapsed / self._slow_start_duration
        multiplier = 0.1 + (0.9 * progress)
        
        return self.weight * multiplier
    
    @property
    def load_ratio(self) -> float:
        """Load ratio using effective weight."""
        ew = self.effective_weight
        if ew == 0:
            return float('inf')
        return self.active_connections / ew
 
 
class SlowStartLeastConnections:
    """
    Least Connections with slow-start support.
    
    New/recovered servers receive gradually increasing traffic
    over the slow-start period, preventing thundering herd.
    """
    
    def __init__(self, servers: dict[str, int], slow_start_seconds: float = 60.0):
        self._slow_start_duration = slow_start_seconds
        self._servers = {}
        self._lock = threading.Lock()
        
        for addr, weight in servers.items():
            server = ServerWithSlowStart(
                address=addr,
                weight=weight,
                healthy=True,
            )
            server._slow_start_duration = slow_start_seconds
            # Existing servers don't need slow start
            server._startup_time = None
            self._servers[addr] = server
    
    def add_server(self, address: str, weight: int = 1) -> None:
        """Add a new server with slow-start."""
        with self._lock:
            server = ServerWithSlowStart(
                address=address,
                weight=weight,
                healthy=False,  # Will be marked healthy with slow-start
            )
            server._slow_start_duration = self._slow_start_duration
            self._servers[address] = server
            
            # Trigger slow-start
            server.mark_healthy()
    
    def get_next_server(self) -> Optional[str]:
        """Select server with lowest load ratio (with slow-start adjustment)."""
        with self._lock:
            healthy = [s for s in self._servers.values() if s.healthy]
            
            if not healthy:
                return None
            
            best = min(healthy, key=lambda s: s.load_ratio)
            best.active_connections += 1
            return best.address
    
    def release_connection(self, address: str) -> None:
        """Release connection when complete."""
        with self._lock:
            if address in self._servers:
                s = self._servers[address]
                s.active_connections = max(0, s.active_connections - 1)

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

import random
import threading
from typing import Optional
 
class P2CLeastConnections:
    """
    Power of Two Random Choices + Least Connections.
    
    Provides O(1) selection with near-optimal load distribution.
    Especially valuable for large server pools (100+ servers).
    """
    
    def __init__(self, servers: list[str]):
        self._servers = list(servers)
        self._connections = {s: 0 for s in servers}
        self._lock = threading.Lock()
    
    def get_next_server(self) -> Optional[str]:
        """
        Select server using P2C algorithm.
        
        Time Complexity: O(1)
        Distribution: Provably near-optimal for large pools
        """
        with self._lock:
            if len(self._servers) < 2:
                if self._servers:
                    server = self._servers[0]
                    self._connections[server] += 1
                    return server
                return None
            
            # Pick two random servers
            a, b = random.sample(self._servers, 2)
            
            # Choose the one with fewer connections
            if self._connections[a] <= self._connections[b]:
                self._connections[a] += 1
                return a
            else:
                self._connections[b] += 1
                return b
    
    def release_connection(self, address: str) -> None:
        """Release connection."""
        with self._lock:
            if address in self._connections:
                self._connections[address] = max(0, self._connections[address] - 1)
 
 
# Benchmark comparison
if __name__ == "__main__":
    import time
    
    # Large server pool
    servers = [f"server-{i}" for i in range(1000)]
    
    # P2C is O(1)
    p2c_lb = P2CLeastConnections(servers)
    
    # Time 100,000 selections
    start = time.perf_counter()
    for _ in range(100_000):
        server = p2c_lb.get_next_server()
        p2c_lb.release_connection(server)
    duration = time.perf_counter() - start
    
    print(f"P2C: 100,000 selections in {duration:.3f}s")
    print(f"     {100_000/duration:.0f} selections/second")