Resource Allocation Graph

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
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237

"""
Resource Allocation Graph - Data Structure Implementation
 
This module provides a rigorous implementation of the Resource Allocation
Graph (RAG) data structure, including support for both single-instance
and multi-instance resources.
"""
 
from dataclasses import dataclass, field
from enum import Enum, auto
from typing import Dict, List, Set, Optional, Tuple
from collections import defaultdict
 
 
class VertexType(Enum):
    """Vertex type in the Resource Allocation Graph."""
    PROCESS = auto()
    RESOURCE = auto()
 
 
class EdgeType(Enum):
    """Edge type in the Resource Allocation Graph."""
    REQUEST = auto()      # Process -> Resource (wants resource)
    ASSIGNMENT = auto()   # Resource -> Process (holds resource)
 
 
@dataclass
class ResourceType:
    """
    Represents a resource type in the system.
    
    Attributes:
        name: Unique identifier for this resource type
        total_instances: Total number of instances of this resource type (k_j)
        allocated_instances: Map from process to count of held instances
    """
    name: str
    total_instances: int
    allocated_instances: Dict[str, int] = field(default_factory=dict)
    
    @property
    def available_instances(self) -> int:
        """Number of currently available instances."""
        return self.total_instances - sum(self.allocated_instances.values())
    
    def is_single_instance(self) -> bool:
        """True if this resource type has exactly one instance."""
        return self.total_instances == 1
 
 
@dataclass
class Process:
    """
    Represents a process in the system.
    
    Attributes:
        name: Unique identifier for this process
        held_resources: Map from resource type to count of held instances
        requested_resources: Map from resource type to count of requested instances
    """
    name: str
    held_resources: Dict[str, int] = field(default_factory=dict)
    requested_resources: Dict[str, int] = field(default_factory=dict)
    
    def is_blocked(self) -> bool:
        """True if this process is waiting for any resource."""
        return any(count > 0 for count in self.requested_resources.values())
 
 
class ResourceAllocationGraph:
    """
    A directed bipartite graph representing resource allocation state.
    
    The graph G = (V, E) where:
    - V = P ∪ R (processes ∪ resources), P ∩ R = ∅
    - E = E_request ∪ E_assignment
    
    This implementation supports:
    - Single and multi-instance resources
    - Request and assignment edge tracking
    - Cycle detection for deadlock analysis
    """
    
    def __init__(self):
        self.processes: Dict[str, Process] = {}
        self.resources: Dict[str, ResourceType] = {}
        
        # Adjacency representation for graph algorithms
        # request_edges[p] = list of resources process p is requesting
        self.request_edges: Dict[str, List[str]] = defaultdict(list)
        # assignment_edges[r] = list of processes holding resource r
        self.assignment_edges: Dict[str, List[str]] = defaultdict(list)
    
    def add_process(self, name: str) -> Process:
        """Add a process vertex to the graph."""
        if name in self.processes:
            raise ValueError(f"Process {name} already exists")
        if name in self.resources:
            raise ValueError(f"Name {name} conflicts with existing resource")
        
        process = Process(name=name)
        self.processes[name] = process
        return process
    
    def add_resource(self, name: str, instances: int = 1) -> ResourceType:
        """Add a resource type vertex to the graph."""
        if instances < 1:
            raise ValueError(f"Resource must have at least 1 instance")
        if name in self.resources:
            raise ValueError(f"Resource {name} already exists")
        if name in self.processes:
            raise ValueError(f"Name {name} conflicts with existing process")
        
        resource = ResourceType(name=name, total_instances=instances)
        self.resources[name] = resource
        return resource
    
    def add_request_edge(self, process: str, resource: str, count: int = 1) -> bool:
        """
        Add a request edge: process -> resource.
        
        Represents that the process is requesting 'count' instances
        of the resource type.
        
        Returns True if the edge was added, False if already exists.
        """
        if process not in self.processes:
            raise ValueError(f"Unknown process: {process}")
        if resource not in self.resources:
            raise ValueError(f"Unknown resource: {resource}")
        
        proc = self.processes[process]
        current_request = proc.requested_resources.get(resource, 0)
        proc.requested_resources[resource] = current_request + count
        
        # Update adjacency list for graph traversal
        for _ in range(count):
            self.request_edges[process].append(resource)
        
        return True
    
    def add_assignment_edge(self, resource: str, process: str, count: int = 1) -> bool:
        """
        Add an assignment edge: resource -> process.
        
        Represents that 'count' instances of the resource type
        are allocated to the process.
        
        Returns True if successful, raises if insufficient instances.
        """
        if process not in self.processes:
            raise ValueError(f"Unknown process: {process}")
        if resource not in self.resources:
            raise ValueError(f"Unknown resource: {resource}")
        
        res = self.resources[resource]
        if res.available_instances < count:
            raise ValueError(
                f"Cannot allocate {count} instances of {resource}: "
                f"only {res.available_instances} available"
            )
        
        # Update resource allocation tracking
        current_alloc = res.allocated_instances.get(process, 0)
        res.allocated_instances[process] = current_alloc + count
        
        # Update process holdings
        proc = self.processes[process]
        current_held = proc.held_resources.get(resource, 0)
        proc.held_resources[resource] = current_held + count
        
        # Update adjacency list
        for _ in range(count):
            self.assignment_edges[resource].append(process)
        
        return True
    
    def get_vertex_count(self) -> Tuple[int, int]:
        """Returns (process_count, resource_count)."""
        return len(self.processes), len(self.resources)
    
    def get_edge_count(self) -> Tuple[int, int]:
        """Returns (request_edge_count, assignment_edge_count)."""
        request_count = sum(len(edges) for edges in self.request_edges.values())
        assign_count = sum(len(edges) for edges in self.assignment_edges.values())
        return request_count, assign_count
    
    def is_bipartite(self) -> bool:
        """
        Verify graph maintains bipartite property.
        
        All edges must connect P to R (never P to P or R to R).
        This is enforced by construction, so always returns True.
        """
        # Request edges: P -> R (by type checking in add_request_edge)
        # Assignment edges: R -> P (by type checking in add_assignment_edge)
        return True
    
    def __repr__(self) -> str:
        p_count, r_count = self.get_vertex_count()
        req_count, asgn_count = self.get_edge_count()
        return (
            f"ResourceAllocationGraph("
            f"processes={p_count}, resources={r_count}, "
            f"request_edges={req_count}, assignment_edges={asgn_count})"
        )
 
 
# Example: Building a simple RAG
if __name__ == "__main__":
    rag = ResourceAllocationGraph()
    
    # Add processes
    rag.add_process("P1")
    rag.add_process("P2")
    rag.add_process("P3")
    
    # Add resources (single instance for simplicity)
    rag.add_resource("R1", instances=1)
    rag.add_resource("R2", instances=1)
    rag.add_resource("R3", instances=2)  # Multi-instance
    
    # P1 holds R1, requests R2
    rag.add_assignment_edge("R1", "P1")
    rag.add_request_edge("P1", "R2")
    
    # P2 holds R2, requests R3
    rag.add_assignment_edge("R2", "P2")
    rag.add_request_edge("P2", "R3")
    
    # P3 holds one instance of R3, requests R1
    rag.add_assignment_edge("R3", "P3")
    rag.add_request_edge("P3", "R1")
    
    print(rag)  # Shows graph statistics
    # Output: ResourceAllocationGraph(processes=3, resources=3, 
    #         request_edges=3, assignment_edges=3)

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

def analyze_resource_instances(rag: ResourceAllocationGraph) -> dict:
    """
    Analyze the instance characteristics of a Resource Allocation Graph.
    
    Returns a dictionary with analysis results including:
    - is_single_instance_only: True if all resources have exactly one instance
    - cycle_implies_deadlock: True if we can use simple cycle detection
    - resource_breakdown: Details for each resource type
    """
    single_instance_resources = []
    multi_instance_resources = []
    
    for name, resource in rag.resources.items():
        if resource.is_single_instance():
            single_instance_resources.append(name)
        else:
            multi_instance_resources.append(name)
    
    is_single_only = len(multi_instance_resources) == 0
    
    return {
        "is_single_instance_only": is_single_only,
        "cycle_implies_deadlock": is_single_only,  # Only true for single-instance!
        "single_instance_resources": single_instance_resources,
        "multi_instance_resources": multi_instance_resources,
        "detection_algorithm": (
            "Simple DFS cycle detection" if is_single_only 
            else "Graph reduction or Banker's algorithm required"
        ),
        "complexity": (
            "O(V + E)" if is_single_only
            else "O(m × n²) where m = resource types, n = processes"
        )
    }
 
 
def illustrate_cycle_difference():
    """
    Demonstrates the difference between cycles in single-instance
    vs multi-instance resource graphs.
    """
    # Example 1: Single-instance - cycle MEANS deadlock
    print("=== Single Instance Example ===")
    rag1 = ResourceAllocationGraph()
    rag1.add_process("P1")
    rag1.add_process("P2")
    rag1.add_resource("R1", instances=1)  # Single instance
    rag1.add_resource("R2", instances=1)  # Single instance
    
    # P1 holds R1, wants R2
    rag1.add_assignment_edge("R1", "P1")
    rag1.add_request_edge("P1", "R2")
    
    # P2 holds R2, wants R1
    rag1.add_assignment_edge("R2", "P2")
    rag1.add_request_edge("P2", "R1")
    
    analysis1 = analyze_resource_instances(rag1)
    print(f"Cycle detection sufficient: {analysis1['cycle_implies_deadlock']}")
    print(f"Algorithm: {analysis1['detection_algorithm']}")
    # Has cycle P1->R2->P2->R1->P1, therefore DEADLOCK
    
    # Example 2: Multi-instance - cycle does NOT mean deadlock
    print("\n=== Multi Instance Example ===")
    rag2 = ResourceAllocationGraph()
    rag2.add_process("P1")
    rag2.add_process("P2")
    rag2.add_process("P3")
    rag2.add_resource("R", instances=2)  # Multi instance!
    
    # P1 holds one R, wants another R
    rag2.add_assignment_edge("R", "P1")
    rag2.add_request_edge("P1", "R")
    
    # P2 holds one R
    rag2.add_assignment_edge("R", "P2")
    # P2 is NOT requesting anything - can finish!
    
    analysis2 = analyze_resource_instances(rag2)
    print(f"Cycle detection sufficient: {analysis2['cycle_implies_deadlock']}")
    print(f"Algorithm: {analysis2['detection_algorithm']}")
    # P1 has cycle on R, but P2 can release, breaking potential deadlock
 
 
if __name__ == "__main__":
    illustrate_cycle_difference()

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
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224

/**
 * Resource Allocation Graph Construction - Kernel Implementation
 * 
 * This module demonstrates how an operating system kernel might
 * construct and maintain a Resource Allocation Graph for deadlock
 * detection purposes.
 */
 
#include <linux/types.h>
#include <linux/list.h>
#include <linux/spinlock.h>
#include <linux/slab.h>
 
/* Forward declarations */
struct process_t;
struct resource_t;
struct rag_edge;
struct resource_allocation_graph;
 
/* Edge types in the RAG */
enum edge_type {
    EDGE_REQUEST,     /* Process -> Resource */
    EDGE_ASSIGNMENT   /* Resource -> Process */
};
 
/* An edge in the Resource Allocation Graph */
struct rag_edge {
    enum edge_type type;
    
    union {
        struct {
            struct process_t *from;
            struct resource_t *to;
        } request;
        struct {
            struct resource_t *from;
            struct process_t *to;
            int instance_id;  /* Which specific instance */
        } assignment;
    };
    
    struct list_head list;  /* For linking in adjacency lists */
};
 
/* The complete Resource Allocation Graph */
struct resource_allocation_graph {
    struct list_head processes;      /* All process vertices */
    struct list_head resources;      /* All resource vertices */
    struct list_head all_edges;      /* All edges for traversal */
    
    spinlock_t lock;                 /* Protects entire graph */
    
    /* Statistics */
    int process_count;
    int resource_count;
    int request_edge_count;
    int assignment_edge_count;
};
 
/* Global RAG instance */
static struct resource_allocation_graph *system_rag;
 
/**
 * Initialize the Resource Allocation Graph
 */
int rag_init(void)
{
    system_rag = kzalloc(sizeof(*system_rag), GFP_KERNEL);
    if (!system_rag)
        return -ENOMEM;
    
    INIT_LIST_HEAD(&system_rag->processes);
    INIT_LIST_HEAD(&system_rag->resources);
    INIT_LIST_HEAD(&system_rag->all_edges);
    spin_lock_init(&system_rag->lock);
    
    return 0;
}
 
/**
 * Add a request edge: Process P is requesting Resource R
 * 
 * Called when a process blocks on a resource request.
 * Must be called with rag->lock held.
 */
int rag_add_request_edge(struct process_t *proc, struct resource_t *res)
{
    struct rag_edge *edge;
    
    edge = kzalloc(sizeof(*edge), GFP_ATOMIC);
    if (!edge)
        return -ENOMEM;
    
    edge->type = EDGE_REQUEST;
    edge->request.from = proc;
    edge->request.to = res;
    
    /* Add to global edge list for traversal */
    list_add_tail(&edge->list, &system_rag->all_edges);
    
    /* Add to process's outgoing edge list */
    list_add_tail(&edge->list, &proc->outgoing_edges);
    
    system_rag->request_edge_count++;
    
    /* This is where we might trigger deadlock check */
    return rag_check_for_cycle_from(proc);
}
 
/**
 * Add an assignment edge: Resource R instance is allocated to Process P
 * 
 * Called when a resource is successfully granted to a process.
 * Must be called with rag->lock held.
 */
int rag_add_assignment_edge(struct resource_t *res, struct process_t *proc, 
                            int instance_id)
{
    struct rag_edge *edge;
    
    edge = kzalloc(sizeof(*edge), GFP_ATOMIC);
    if (!edge)
        return -ENOMEM;
    
    edge->type = EDGE_ASSIGNMENT;
    edge->assignment.from = res;
    edge->assignment.to = proc;
    edge->assignment.instance_id = instance_id;
    
    list_add_tail(&edge->list, &system_rag->all_edges);
    list_add_tail(&edge->list, &res->outgoing_edges);
    
    system_rag->assignment_edge_count++;
    
    return 0;
}
 
/**
 * Remove a request edge: Process P received its requested resource
 * 
 * Called when a blocked process is granted its resource.
 * The request edge is replaced by an assignment edge.
 */
void rag_remove_request_edge(struct process_t *proc, struct resource_t *res)
{
    struct rag_edge *edge, *tmp;
    
    list_for_each_entry_safe(edge, tmp, &proc->outgoing_edges, list) {
        if (edge->type == EDGE_REQUEST && 
            edge->request.to == res) {
            list_del(&edge->list);
            system_rag->request_edge_count--;
            kfree(edge);
            return;
        }
    }
}
 
/**
 * Remove an assignment edge: Process P released resource R
 * 
 * Called when a process releases a resource.
 */
void rag_remove_assignment_edge(struct resource_t *res, struct process_t *proc)
{
    struct rag_edge *edge, *tmp;
    
    list_for_each_entry_safe(edge, tmp, &res->outgoing_edges, list) {
        if (edge->type == EDGE_ASSIGNMENT && 
            edge->assignment.to == proc) {
            list_del(&edge->list);
            system_rag->assignment_edge_count--;
            kfree(edge);
            return;
        }
    }
}
 
/**
 * Build complete RAG snapshot from current system state
 * 
 * This reconstructs the entire graph by iterating over all
 * processes and resources. Used for verification or when
 * incremental maintenance is not possible.
 */
int rag_build_from_system_state(void)
{
    struct task_struct *task;
    struct resource_t *res;
    unsigned long flags;
    
    spin_lock_irqsave(&system_rag->lock, flags);
    
    /* Clear existing graph */
    rag_clear_all_edges();
    
    /* Phase 1 & 2: Vertices are maintained separately */
    
    /* Phase 3: Add assignment edges from allocation tables */
    list_for_each_entry(res, &system_rag->resources, list) {
        for (int i = 0; i < res->total_instances; i++) {
            if (res->instances[i].holder != NULL) {
                rag_add_assignment_edge(
                    res, 
                    res->instances[i].holder,
                    i
                );
            }
        }
    }
    
    /* Phase 4: Add request edges from wait queues */
    list_for_each_entry(res, &system_rag->resources, list) {
        struct wait_queue_entry *wqe;
        list_for_each_entry(wqe, &res->wait_queue, list) {
            struct process_t *waiting = wqe->private;
            rag_add_request_edge(waiting, res);
        }
    }
    
    spin_unlock_irqrestore(&system_rag->lock, flags);
    
    return 0;
}

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

"""
Resource Allocation Graph Invariant Verification
 
This module provides comprehensive verification of RAG invariants,
useful for debugging, testing, and runtime consistency checks.
"""
 
from typing import List, Tuple, Optional
from dataclasses import dataclass
 
 
@dataclass
class InvariantViolation:
    """Represents a detected invariant violation."""
    invariant_name: str
    description: str
    severity: str  # "ERROR", "WARNING"
    affected_vertices: List[str]
    
 
class RAGInvariantChecker:
    """
    Verifies that a Resource Allocation Graph satisfies all
    required invariants for correctness.
    """
    
    def __init__(self, rag: 'ResourceAllocationGraph'):
        self.rag = rag
        self.violations: List[InvariantViolation] = []
    
    def check_all_invariants(self) -> List[InvariantViolation]:
        """Run all invariant checks and return any violations."""
        self.violations = []
        
        self._check_bipartiteness()
        self._check_direction_consistency()
        self._check_instance_constraints()
        self._check_assignment_accounting()
        self._check_no_self_loops()
        
        return self.violations
    
    def _check_bipartiteness(self):
        """
        Invariant 1: Graph must be bipartite.
        All edges connect P to R, never P-P or R-R.
        """
        for process_name, edges in self.rag.request_edges.items():
            # Request edges: should go from process to resource
            if process_name not in self.rag.processes:
                self.violations.append(InvariantViolation(
                    invariant_name="Bipartiteness (Request)",
                    description=f"Request edge source '{process_name}' is not a process",
                    severity="ERROR",
                    affected_vertices=[process_name]
                ))
            for resource_name in edges:
                if resource_name not in self.rag.resources:
                    self.violations.append(InvariantViolation(
                        invariant_name="Bipartiteness (Request)",
                        description=f"Request edge target '{resource_name}' is not a resource",
                        severity="ERROR",
                        affected_vertices=[process_name, resource_name]
                    ))
        
        for resource_name, processes in self.rag.assignment_edges.items():
            # Assignment edges: should go from resource to process
            if resource_name not in self.rag.resources:
                self.violations.append(InvariantViolation(
                    invariant_name="Bipartiteness (Assignment)",
                    description=f"Assignment edge source '{resource_name}' is not a resource",
                    severity="ERROR",
                    affected_vertices=[resource_name]
                ))
            for process_name in processes:
                if process_name not in self.rag.processes:
                    self.violations.append(InvariantViolation(
                        invariant_name="Bipartiteness (Assignment)",
                        description=f"Assignment edge target '{process_name}' is not a process",
                        severity="ERROR",
                        affected_vertices=[resource_name, process_name]
                    ))
    
    def _check_direction_consistency(self):
        """
        Invariant 2: Edge directions are semantically correct.
        Request: P → R
        Assignment: R → P
        """
        # This is enforced by the data structure design
        # Request edges are stored as process -> [resources]
        # Assignment edges are stored as resource -> [processes]
        # Verify the structure matches the semantics
        pass  # Checked implicitly by bipartiteness
    
    def _check_instance_constraints(self):
        """
        Invariant 3: Cannot allocate more instances than exist.
        |assignment edges from R| ≤ |R| for each resource R.
        """
        for resource_name, resource in self.rag.resources.items():
            assigned_count = len(self.rag.assignment_edges.get(resource_name, []))
            total_instances = resource.total_instances
            
            if assigned_count > total_instances:
                self.violations.append(InvariantViolation(
                    invariant_name="Instance Constraint",
                    description=(
                        f"Resource '{resource_name}' has {assigned_count} "
                        f"assignment edges but only {total_instances} instances"
                    ),
                    severity="ERROR",
                    affected_vertices=[resource_name]
                ))
    
    def _check_assignment_accounting(self):
        """
        Invariant 7: Instance accounting must balance.
        assigned + available = total for each resource.
        """
        for resource_name, resource in self.rag.resources.items():
            assigned = sum(resource.allocated_instances.values())
            available = resource.available_instances
            total = resource.total_instances
            
            if assigned + available != total:
                self.violations.append(InvariantViolation(
                    invariant_name="Assignment Accounting",
                    description=(
                        f"Resource '{resource_name}': "
                        f"assigned({assigned}) + available({available}) "
                        f"!= total({total})"
                    ),
                    severity="ERROR",
                    affected_vertices=[resource_name]
                ))
    
    def _check_no_self_loops(self):
        """
        Additional invariant: No self-loops should exist.
        A vertex should not have an edge to itself.
        """
        # In a proper bipartite graph between P and R, self-loops
        # are impossible by construction. This is a sanity check.
        for process_name in self.rag.processes:
            if process_name in self.rag.request_edges.get(process_name, []):
                self.violations.append(InvariantViolation(
                    invariant_name="No Self-Loops",
                    description=f"Process '{process_name}' has edge to itself",
                    severity="ERROR",
                    affected_vertices=[process_name]
                ))
    
    def is_valid(self) -> bool:
        """Returns True if no ERROR-level violations exist."""
        violations = self.check_all_invariants()
        return not any(v.severity == "ERROR" for v in violations)
    
    def print_report(self):
        """Print a human-readable invariant check report."""
        violations = self.check_all_invariants()
        
        if not violations:
            print("✓ All RAG invariants satisfied")
            return
        
        print(f"✗ Found {len(violations)} invariant violation(s):\n")
        
        for v in violations:
            severity_symbol = "ERROR" if v.severity == "ERROR" else "WARN"
            print(f"[{severity_symbol}] {v.invariant_name}")
            print(f"        {v.description}")
            print(f"        Affected: {', '.join(v.affected_vertices)}\n")
 
 
# Example usage
def demonstrate_invariant_checking():
    """Demonstrate the invariant checker with a sample RAG."""
    rag = ResourceAllocationGraph()
    
    # Build a valid graph
    rag.add_process("P1")
    rag.add_process("P2")
    rag.add_resource("R1", instances=2)
    rag.add_resource("R2", instances=1)
    
    rag.add_assignment_edge("R1", "P1")
    rag.add_assignment_edge("R2", "P2")
    rag.add_request_edge("P1", "R2")
    rag.add_request_edge("P2", "R1")
    
    # Check invariants
    checker = RAGInvariantChecker(rag)
    checker.print_report()
    
    print(f"\nGraph is valid: {checker.is_valid()}")
 
 
if __name__ == "__main__":
    demonstrate_invariant_checking()