Priority Inversion

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

"""
Stack Resource Policy (SRP) Conceptual Implementation
 
Demonstrates key SRP concepts: preemption levels, system ceiling,
and early blocking at task activation (not resource acquisition).
"""
 
from dataclasses import dataclass, field
from typing import Dict, List, Optional
 
@dataclass
class Resource:
    """SRP Resource with preemption ceiling."""
    id: str
    # Maximum units available (supports counting resources)
    max_units: int
    available_units: int
    # Preemption ceiling = max preemption level of any task using this
    ceiling: int
 
@dataclass
class Task:
    """SRP Task with priority and preemption level."""
    id: str
    priority: int           # For scheduling (lower = higher priority)
    preemption_level: int   # For resource access (lower = can preempt more)
    # Maximum resources this task might hold simultaneously
    max_resource_usage: Dict[str, int] = field(default_factory=dict)
 
class SRPScheduler:
    """
    Stack Resource Policy scheduler demonstrating core concepts.
    
    Key invariant: A task never blocks after it starts executing.
    This enables stack sharing among all tasks.
    """
    
    def __init__(self):
        self.resources: Dict[str, Resource] = {}
        self.tasks: Dict[str, Task] = {}
        self.held_resources: Dict[str, Dict[str, int]] = {}  # task -> resource -> count
        self.system_ceiling: int = 0  # Lower = more restrictive
    
    def compute_resource_ceilings(self):
        """
        Compute resource ceilings based on task usage.
        Ceiling = min preemption level of any task using the resource.
        """
        for res_id, resource in self.resources.items():
            ceiling = float('inf')
            for task_id, task in self.tasks.items():
                if res_id in task.max_resource_usage:
                    ceiling = min(ceiling, task.preemption_level)
            resource.ceiling = ceiling if ceiling != float('inf') else 0
            print(f"Resource {res_id} ceiling = {resource.ceiling}")
    
    def update_system_ceiling(self):
        """
        System ceiling = min ceiling among all currently held resources.
        A task can only start/resume if its preemption level <= system ceiling.
        """
        ceilings = []
        for task_id, held in self.held_resources.items():
            for res_id, count in held.items():
                if count > 0:
                    ceilings.append(self.resources[res_id].ceiling)
        
        self.system_ceiling = min(ceilings) if ceilings else float('inf')
    
    def can_preempt(self, task_id: str) -> bool:
        """
        Check if task can start/resume execution.
        SRP rule: Task can run iff preemption_level < system_ceiling.
        """
        task = self.tasks[task_id]
        can = task.preemption_level < self.system_ceiling
        print(f"Task {task_id} (pl={task.preemption_level}) "
              f"vs ceiling {self.system_ceiling}: "
              f"{'CAN' if can else 'CANNOT'} preempt")
        return can
    
    def get_resource(self, task_id: str, resource_id: str, 
                     count: int = 1) -> bool:
        """
        Acquire resource units. Under SRP invariants, this should
        never block because early blocking prevents the scenario.
        """
        resource = self.resources[resource_id]
        if resource.available_units >= count:
            resource.available_units -= count
            if task_id not in self.held_resources:
                self.held_resources[task_id] = {}
            self.held_resources[task_id][resource_id] =                 self.held_resources[task_id].get(resource_id, 0) + count
            self.update_system_ceiling()
            return True
        else:
            # Should never happen under correct SRP - indicates bug
            raise RuntimeError(f"SRP violation: {task_id} blocked on {resource_id}")
    
    def release_resource(self, task_id: str, resource_id: str, 
                         count: int = 1):
        """Release resource units."""
        resource = self.resources[resource_id]
        resource.available_units += count
        self.held_resources[task_id][resource_id] -= count
        if self.held_resources[task_id][resource_id] == 0:
            del self.held_resources[task_id][resource_id]
        self.update_system_ceiling()
 
 
def demonstrate_srp():
    """
    Demonstrate SRP with counting resources.
    
    Scenario:
    - Resource pool with 3 units
    - τ1 (pri=1, pl=1) needs 2 units
    - τ2 (pri=2, pl=2) needs 2 units
    - τ3 (pri=3, pl=3) needs 1 unit
    
    SRP ensures τ3 cannot start while τ2 holds 2 units,
    preventing scenario where τ1 couldn't get 2 units.
    """
    sched = SRPScheduler()
    
    # Setup
    sched.resources = {
        'pool': Resource('pool', max_units=3, available_units=3, ceiling=0)
    }
    sched.tasks = {
        'τ1': Task('τ1', priority=1, preemption_level=1, 
                   max_resource_usage={'pool': 2}),
        'τ2': Task('τ2', priority=2, preemption_level=2,
                   max_resource_usage={'pool': 2}),
        'τ3': Task('τ3', priority=3, preemption_level=3,
                   max_resource_usage={'pool': 1}),
    }
    
    sched.compute_resource_ceilings()
    print()
    
    print("=== Scenario: τ3 holds 1 unit, τ2 wants to start ===")
    sched.get_resource('τ3', 'pool', 1)
    print(f"τ3 holds 1 unit, available={sched.resources['pool'].available_units}")
    print(f"System ceiling = {sched.system_ceiling}")
    
    print()
    print("τ2 (pl=2) tries to preempt:")
    sched.can_preempt('τ2')
    
    print()
    print("τ1 (pl=1) tries to preempt:")
    sched.can_preempt('τ1')
 
 
if __name__ == "__main__":
    demonstrate_srp()

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

/**
 * Resource Protocol Configuration Across RTOSes
 * 
 * Shows how to configure resource access protocols in major RTOSes.
 */
 
/* ============================================
 * POSIX (Linux, QNX, RTEMS POSIX layer)
 * ============================================ */
#include <pthread.h>
 
void posix_pip_mutex(pthread_mutex_t* mutex) {
    pthread_mutexattr_t attr;
    pthread_mutexattr_init(&attr);
    pthread_mutexattr_setprotocol(&attr, PTHREAD_PRIO_INHERIT);
    pthread_mutex_init(mutex, &attr);
    pthread_mutexattr_destroy(&attr);
}
 
void posix_pcp_mutex(pthread_mutex_t* mutex, int ceiling) {
    pthread_mutexattr_t attr;
    pthread_mutexattr_init(&attr);
    pthread_mutexattr_setprotocol(&attr, PTHREAD_PRIO_PROTECT);
    pthread_mutexattr_setprioceiling(&attr, ceiling);
    pthread_mutex_init(mutex, &attr);
    pthread_mutexattr_destroy(&attr);
}
 
/* ============================================
 * VxWorks
 * ============================================ */
#ifdef VXWORKS
#include <semLib.h>
 
SEM_ID vxworks_pip_mutex(void) {
    /* SEM_INVERSION_SAFE enables priority inheritance */
    return semMCreate(SEM_Q_PRIORITY | SEM_INVERSION_SAFE);
}
 
SEM_ID vxworks_regular_mutex(void) {
    /* Without SEM_INVERSION_SAFE - vulnerable to inversion! */
    return semMCreate(SEM_Q_PRIORITY);
}
#endif
 
/* ============================================
 * FreeRTOS
 * ============================================ */
#ifdef FREERTOS
#include "FreeRTOS.h"
#include "semphr.h"
 
SemaphoreHandle_t freertos_pip_mutex(void) {
    /* xSemaphoreCreateMutex automatically uses PIP */
    return xSemaphoreCreateMutex();
}
 
SemaphoreHandle_t freertos_recursive_mutex(void) {
    /* Recursive mutex also has PIP */
    return xSemaphoreCreateRecursiveMutex();
}
 
/* Note: FreeRTOS does not provide built-in PCP.
 * For PCP, implement ceiling check in application:
 * 
 * if (current_priority > mutex_ceiling) {
 *     block_until_ceiling_allows();
 * }
 * vTaskPrioritySet(current_task, mutex_ceiling);
 * xSemaphoreTake(mutex, portMAX_DELAY);
 */
#endif
 
/* ============================================
 * Zephyr RTOS
 * ============================================ */
#ifdef ZEPHYR
#include <zephyr/kernel.h>
 
/* Zephyr k_mutex has automatic priority inheritance */
struct k_mutex zephyr_mutex;
 
void zephyr_init_mutex(void) {
    k_mutex_init(&zephyr_mutex);
    /* PIP is automatic for k_mutex */
}
 
/* For priority ceiling in Zephyr (if CONFIG_PRIORITY_CEILING=y),
 * set ceiling via:
 * k_mutex_ceiling_set(&mutex, ceiling_priority);
 */
#endif

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

/**
 * Resource Server Pattern
 * 
 * A dedicated task owns a shared resource and handles
 * all access requests, eliminating priority inversion.
 */
 
#include "freertos/FreeRTOS.h"
#include "freertos/queue.h"
 
/* Request types for the resource server */
typedef enum {
    REQ_READ,
    REQ_WRITE,
    REQ_CONFIG
} request_type_t;
 
typedef struct {
    request_type_t type;
    void* data;
    size_t length;
    TaskHandle_t requester;
} resource_request_t;
 
/* Response sent back to requester */
typedef struct {
    int status;
    void* data;
    size_t length;
} resource_response_t;
 
static QueueHandle_t request_queue;
static QueueHandle_t response_queue;
 
/**
 * Resource server task - owns the shared hardware.
 * Runs at priority higher than or equal to all clients.
 */
void resource_server_task(void* param) {
    resource_request_t request;
    resource_response_t response;
    
    /* Initialize the actual hardware resource */
    hardware_resource_init();
    
    for (;;) {
        /* Wait for client requests */
        if (xQueueReceive(request_queue, &request, portMAX_DELAY)) {
            
            /* Process request - no locking needed, we're the only accessor */
            switch (request.type) {
                case REQ_READ:
                    response.status = hardware_read(request.data, 
                                                    request.length);
                    break;
                case REQ_WRITE:
                    response.status = hardware_write(request.data,
                                                     request.length);
                    break;
                case REQ_CONFIG:
                    response.status = hardware_config(request.data);
                    break;
            }
            
            /* Send response back to requesting task */
            xQueueSend(response_queue, &response, portMAX_DELAY);
        }
    }
}
 
/**
 * Client API - any task can call this, any priority.
 * No priority inversion possible because no shared locks.
 */
int resource_write(void* data, size_t length) {
    resource_request_t request = {
        .type = REQ_WRITE,
        .data = data,
        .length = length,
        .requester = xTaskGetCurrentTaskHandle()
    };
    resource_response_t response;
    
    /* Send request to server */
    xQueueSend(request_queue, &request, portMAX_DELAY);
    
    /* Wait for response */
    xQueueReceive(response_queue, &response, portMAX_DELAY);
    
    return response.status;
}