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

from collections import defaultdict
from typing import Optional
from enum import Enum
 
class Color(Enum):
    WHITE = 0
    GRAY = 1
    BLACK = 2
 
 
class DeadlockDetector:
    """
    Detects deadlocks using wait-for graph cycle detection.
    """
    
    def __init__(self):
        # Resource held by which process: resource_id -> process_id
        self.resource_holder: dict[str, str] = {}
        # Resources requested by process: process_id -> set of resource_ids
        self.waiting_for: dict[str, set[str]] = defaultdict(set)
        # All known processes
        self.processes: set[str] = set()
    
    def acquire(self, process: str, resource: str) -> bool:
        """
        Process attempts to acquire resource.
        Returns True if successful, False if must wait.
        """
        self.processes.add(process)
        
        if resource not in self.resource_holder:
            # Resource is free
            self.resource_holder[resource] = process
            self.waiting_for[process].discard(resource)
            return True
        elif self.resource_holder[resource] == process:
            # Already holds it
            return True
        else:
            # Must wait
            self.waiting_for[process].add(resource)
            return False
    
    def release(self, process: str, resource: str) -> None:
        """Process releases resource."""
        if self.resource_holder.get(resource) == process:
            del self.resource_holder[resource]
    
    def _build_wait_for_graph(self) -> dict[str, list[str]]:
        """
        Build wait-for graph from current state.
        Edge P1 -> P2 means P1 is waiting for resource held by P2.
        """
        graph: dict[str, list[str]] = {p: [] for p in self.processes}
        
        for process, resources in self.waiting_for.items():
            for resource in resources:
                if resource in self.resource_holder:
                    holder = self.resource_holder[resource]
                    if holder != process:
                        graph[process].append(holder)
        
        return graph
    
    def detect_deadlock(self) -> Optional[list[str]]:
        """
        Detect deadlock by finding cycle in wait-for graph.
        Returns the cycle (list of processes) if deadlock exists.
        """
        graph = self._build_wait_for_graph()
        color = {p: Color.WHITE for p in self.processes}
        parent = {p: None for p in self.processes}
        
        def dfs(node: str) -> Optional[list[str]]:
            color[node] = Color.GRAY
            
            for neighbor in graph.get(node, []):
                if color[neighbor] == Color.GRAY:
                    # Cycle found - reconstruct
                    cycle = [neighbor, node]
                    current = node
                    while parent[current] and parent[current] != neighbor:
                        current = parent[current]
                        cycle.append(current)
                    cycle.append(neighbor)
                    return cycle[::-1]
                elif color[neighbor] == Color.WHITE:
                    parent[neighbor] = node
                    result = dfs(neighbor)
                    if result:
                        return result
            
            color[node] = Color.BLACK
            return None
        
        for process in self.processes:
            if color[process] == Color.WHITE:
                cycle = dfs(process)
                if cycle:
                    return cycle
        return None
 
 
# Example: Classic dining philosophers deadlock
detector = DeadlockDetector()
 
# Each philosopher tries to pick up left fork, then right fork
detector.acquire("P1", "Fork1")  # P1 gets Fork1
detector.acquire("P2", "Fork2")  # P2 gets Fork2
detector.acquire("P1", "Fork2")  # P1 waits for Fork2 (held by P2)
detector.acquire("P2", "Fork1")  # P2 waits for Fork1 (held by P1)
 
deadlock = detector.detect_deadlock()
print(f"Deadlock detected: {deadlock}")  # ['P1', 'P2', 'P1']

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

from collections import defaultdict, deque
from typing import Optional
 
class DependencyResolver:
    """
    Resolves package dependencies and detects circular dependencies.
    """
    
    def __init__(self):
        self.dependencies: dict[str, list[str]] = defaultdict(list)
    
    def add_package(self, package: str, deps: list[str]) -> None:
        """Add package with its dependencies."""
        self.dependencies[package] = deps
        for dep in deps:
            if dep not in self.dependencies:
                self.dependencies[dep] = []
    
    def resolve(self) -> tuple[Optional[list[str]], Optional[list[str]]]:
        """
        Returns (install_order, None) if resolvable,
        or (None, cycle) if circular dependency exists.
        """
        packages = list(self.dependencies.keys())
        in_degree = {p: 0 for p in packages}
        
        # Build reverse graph and compute in-degrees
        for package, deps in self.dependencies.items():
            for dep in deps:
                in_degree[package] += 1
        
        # Start with packages that have no dependencies
        queue = deque([p for p in packages if in_degree[p] == 0])
        install_order = []
        
        while queue:
            package = queue.popleft()
            install_order.append(package)
            
            # Packages that depend on this one
            for pkg, deps in self.dependencies.items():
                if package in deps:
                    in_degree[pkg] -= 1
                    if in_degree[pkg] == 0:
                        queue.append(pkg)
        
        if len(install_order) == len(packages):
            return install_order, None
        
        # Find cycle among remaining packages
        remaining = [p for p in packages if in_degree[p] > 0]
        cycle = self._find_cycle(remaining)
        return None, cycle
    
    def _find_cycle(self, packages: list[str]) -> list[str]:
        """Find a cycle among the given packages."""
        WHITE, GRAY, BLACK = 0, 1, 2
        color = {p: WHITE for p in packages}
        parent = {}
        
        def dfs(node: str) -> Optional[list[str]]:
            color[node] = GRAY
            for dep in self.dependencies[node]:
                if dep in color:
                    if color[dep] == GRAY:
                        cycle = [dep, node]
                        curr = node
                        while parent.get(curr) and parent[curr] != dep:
                            curr = parent[curr]
                            cycle.append(curr)
                        return cycle[::-1]
                    elif color[dep] == WHITE:
                        parent[dep] = node
                        result = dfs(dep)
                        if result:
                            return result
            color[node] = BLACK
            return None
        
        for pkg in packages:
            if color[pkg] == WHITE:
                cycle = dfs(pkg)
                if cycle:
                    return cycle
        return []
 
 
# Example
resolver = DependencyResolver()
resolver.add_package("webapp", ["react", "utils"])
resolver.add_package("react", ["prop-types"])
resolver.add_package("utils", ["lodash"])
resolver.add_package("lodash", [])
resolver.add_package("prop-types", [])
 
order, cycle = resolver.resolve()
print(f"Install order: {order}")  # Valid order
 
# Add circular dependency
resolver.add_package("core", ["webapp"])  # webapp depends on utils
resolver.dependencies["utils"].append("core")  # Creates cycle
 
order, cycle = resolver.resolve()
print(f"Circular dependency: {cycle}")

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

class BuildSystem:
    """Simplified build system with cycle detection."""
    
    def __init__(self):
        self.targets: dict[str, list[str]] = {}
    
    def add_target(self, name: str, deps: list[str]) -> None:
        self.targets[name] = deps
    
    def validate_and_order(self) -> tuple[bool, list[str]]:
        """
        Validate build graph and return build order.
        Returns (success, order_or_cycle).
        """
        WHITE, GRAY, BLACK = 0, 1, 2
        color = {t: WHITE for t in self.targets}
        order = []
        
        def dfs(target: str) -> bool:
            color[target] = GRAY
            for dep in self.targets.get(target, []):
                if dep not in color:
                    continue  # External dependency
                if color[dep] == GRAY:
                    return False  # Cycle!
                if color[dep] == WHITE and not dfs(dep):
                    return False
            color[target] = BLACK
            order.append(target)
            return True
        
        for target in self.targets:
            if color[target] == WHITE:
                if not dfs(target):
                    return False, []
        
        return True, order  # order is in dependency-first order
 
 
# Example: C++ project
build = BuildSystem()
build.add_target("main.o", ["utils.o", "network.o"])
build.add_target("utils.o", ["common.o"])
build.add_target("network.o", ["common.o"])
build.add_target("common.o", [])
 
valid, order = build.validate_and_order()
print(f"Build order: {order}")  # ['common.o', 'utils.o', 'network.o', 'main.o']