Starvation in Concurrency Control

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

class FairLockManager:
    """
    Lock manager with guaranteed starvation freedom.
    
    Properties:
    - Bounded bypass: max K requests can pass waiting request
    - Priority with aging: effective priority increases with wait time
    - Timeout enforcement: no wait exceeds max_timeout
    - Position preservation: retried transactions keep original position
    """
    
    # Configuration
    MAX_BYPASS = 10
    AGING_RATE = 1.0  # priority units per second
    DEFAULT_TIMEOUT = 30_seconds
    
    # State
    lock_tables = ConcurrentHashMap<ResourceId, LockTable>()
    ticket_counter = AtomicLong(0)
    
    def request_lock(transaction, resource, lock_type, timeout=None):
        timeout = timeout or DEFAULT_TIMEOUT
        table = lock_tables.get_or_create(resource)
        
        # Determine ticket number
        if transaction.has_preserved_ticket(resource):
            ticket = transaction.get_preserved_ticket(resource)
        else:
            ticket = ticket_counter.increment_and_get()
        
        request = LockRequest(
            transaction=transaction,
            resource=resource,
            type=lock_type,
            ticket=ticket,
            base_priority=transaction.base_priority,
            request_time=now(),
            bypass_count=0
        )
        
        # Try to grant immediately
        if try_grant_immediately(table, request):
            return LockGranted(request)
        
        # Add to wait queue
        table.wait_queue.add(request)
        
        # Wait with timeout
        deadline = now() + timeout
        
        while now() < deadline:
            if request.granted:
                return LockGranted(request)
            
            wait_until(deadline, or_until=request.granted_signal)
        
        # Timeout - preserve ticket for potential retry
        table.wait_queue.remove(request)
        transaction.preserve_ticket(resource, ticket)
        
        raise LockTimeoutError(resource, timeout)
    
    def try_grant_immediately(table, request):
        """Attempt to grant without waiting"""
        current_holders = table.holders
        wait_queue = table.wait_queue
        
        # Check compatibility with current holders
        for holder in current_holders:
            if conflicts(holder.type, request.type):
                return False
        
        # Check if anyone with better position is waiting
        for waiter in wait_queue:
            if should_serve_before(waiter, request):
                return False
        
        # Grant immediately
        table.holders.add(request)
        request.granted = True
        return True
    
    def should_serve_before(existing, new_request):
        """Determine if existing request should be served before new_request"""
        # Conflicting lock types?
        if not conflicts(existing.type, new_request.type):
            return False  # Compatible, can serve in parallel
        
        # Compare effective priority (includes aging)
        existing_priority = effective_priority(existing)
        new_priority = effective_priority(new_request)
        
        if existing_priority > new_priority:
            return True
        if existing_priority < new_priority:
            # Check bypass limit
            if existing.bypass_count >= MAX_BYPASS:
                return True  # Existing has been bypassed enough
            existing.bypass_count += 1
            return False
        
        # Equal priority - use ticket order (FCFS)
        return existing.ticket < new_request.ticket
    
    def effective_priority(request):
        """Calculate priority with aging"""
        wait_time = now() - request.request_time
        age_bonus = wait_time.seconds * AGING_RATE
        return request.base_priority + age_bonus
    
    def release_lock(transaction, resource):
        table = lock_tables.get(resource)
        if not table:
            return
        
        # Remove from holders
        table.holders.remove_by_transaction(transaction)
        
        # Grant to highest-priority waiters
        grant_waiting_requests(table)
    
    def grant_waiting_requests(table):
        """Grant locks to eligible waiters after a release"""
        # Sort by effective priority, then ticket
        sorted_waiters = sorted(
            table.wait_queue,
            key=lambda r: (-effective_priority(r), r.ticket)
        )
        
        for waiter in sorted_waiters:
            if can_grant(table, waiter):
                table.wait_queue.remove(waiter)
                table.holders.add(waiter)
                waiter.granted = True
                waiter.granted_signal.notify()

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

-- PostgreSQL: Lock wait monitoring dashboard
 
-- Current lock waits
SELECT 
    blocked.pid AS blocked_pid,
    blocked.usename AS blocked_user,
    blocked.query AS blocked_query,
    blocking.pid AS blocking_pid,
    blocking.usename AS blocking_user,
    blocking.query AS blocking_query,
    NOW() - blocked.query_start AS wait_duration
FROM pg_stat_activity blocked
JOIN pg_locks blocked_locks ON blocked.pid = blocked_locks.pid
JOIN pg_locks blocking_locks ON blocked_locks.locktype = blocking_locks.locktype
    AND blocked_locks.database = blocking_locks.database
    AND blocked_locks.relation = blocking_locks.relation
    AND blocked_locks.page = blocking_locks.page
    AND blocked_locks.tuple = blocking_locks.tuple
    AND blocked_locks.transactionid = blocking_locks.transactionid
    AND blocked_locks.classid = blocking_locks.classid
    AND blocked_locks.objid = blocking_locks.objid
    AND blocked_locks.objsubid = blocking_locks.objsubid
    AND blocked_locks.pid != blocking_locks.pid
JOIN pg_stat_activity blocking ON blocking_locks.pid = blocking.pid
WHERE NOT blocked_locks.granted
ORDER BY wait_duration DESC;
 
-- Lock wait time percentiles (requires pg_stat_statements extension)
-- Use application-level logging for accurate tracking
 
-- Alert query: transactions waiting > 10 seconds
SELECT COUNT(*) AS long_waiters
FROM pg_stat_activity
WHERE wait_event_type = 'Lock'
  AND NOW() - query_start > INTERVAL '10 seconds';
 
-- MySQL: Lock monitoring
SELECT * FROM performance_schema.events_waits_summary_by_event_name
WHERE event_name LIKE 'wait/synch/mutex/innodb%'
ORDER BY sum_timer_wait DESC;
 
SELECT waiting_pid, waiting_query, blocking_pid, blocking_query
FROM sys.innodb_lock_waits;

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

-- Step 1: Identify affected transactions
SELECT application_name, state, wait_event_type, wait_event,
       COUNT(*) as count,
       AVG(EXTRACT(EPOCH FROM (now() - query_start))) as avg_wait_sec
FROM pg_stat_activity
WHERE state = 'active' OR wait_event_type = 'Lock'
GROUP BY 1, 2, 3, 4
ORDER BY count DESC;
 
-- Result:
-- application_name | state  | wait_event_type | wait_event | count | avg_wait_sec
-- checkout_service | active | Lock            | tuple      | 47    | 18.3
-- analytics_batch  | active | NULL            | NULL       | 3     | 145.7
 
-- Step 2: Identify the blocker
SELECT blocked.pid, blocked.application_name, 
       blocking.pid as blocking_pid, blocking.application_name as blocking_app,
       blocking.query as blocking_query
FROM pg_stat_activity blocked
JOIN pg_locks bl ON blocked.pid = bl.pid AND NOT bl.granted
JOIN pg_locks holding ON bl.relation = holding.relation AND holding.granted
JOIN pg_stat_activity blocking ON holding.pid = blocking.pid
WHERE blocked.application_name = 'checkout_service';
 
-- Result: analytics_batch is holding locks on inventory table