Applications: Task Scheduling & Build Systems

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

from collections import deque
from dataclasses import dataclass, field
from typing import Callable, Any, Optional
from enum import Enum
import time
 
class TaskStatus(Enum):
    PENDING = "pending"
    RUNNING = "running"
    COMPLETED = "completed"
    FAILED = "failed"
    SKIPPED = "skipped"
 
@dataclass
class Task:
    """Represents a schedulable task with dependencies."""
    name: str
    action: Callable[[], Any]  # The actual work to perform
    dependencies: list[str] = field(default_factory=list)
    status: TaskStatus = TaskStatus.PENDING
    result: Any = None
    error: Optional[Exception] = None
    duration: float = 0.0
 
class TaskScheduler:
    """
    A dependency-aware task scheduler using topological sorting.
    
    Features:
    - Cycle detection with detailed error messages
    - Task execution with timing and error handling
    - Support for skipping tasks dependent on failures
    """
    
    def __init__(self):
        self.tasks: dict[str, Task] = {}
    
    def add_task(self, name: str, action: Callable, dependencies: list[str] = None):
        """Add a task to the scheduler."""
        deps = dependencies or []
        self.tasks[name] = Task(name=name, action=action, dependencies=deps)
    
    def _compute_in_degrees(self) -> dict[str, int]:
        """Compute in-degree for each task."""
        in_degree = {name: 0 for name in self.tasks}
        for task in self.tasks.values():
            for dep in task.dependencies:
                if dep not in self.tasks:
                    raise ValueError(f"Task '{task.name}' depends on unknown task '{dep}'")
                in_degree[task.name] = in_degree.get(task.name, 0) + 1
        # Wait, this is wrong - we need to count incoming edges
        # Let me fix this
        in_degree = {name: 0 for name in self.tasks}
        for task in self.tasks.values():
            for dep in task.dependencies:
                # dep → task.name edge, so task.name's in-degree increases
                pass  # Actually the dependency list IS incoming edges
        # Each task's dependencies ARE its incoming edges
        for name in self.tasks:
            in_degree[name] = len(self.tasks[name].dependencies)
        return in_degree
    
    def _build_adjacency(self) -> dict[str, list[str]]:
        """Build adjacency list (dependency → dependent)."""
        adj = {name: [] for name in self.tasks}
        for task in self.tasks.values():
            for dep in task.dependencies:
                adj[dep].append(task.name)
        return adj
    
    def get_execution_order(self) -> list[str]:
        """
        Compute valid execution order using Kahn's algorithm.
        Raises ValueError if cycle is detected.
        """
        in_degree = self._compute_in_degrees()
        adj = self._build_adjacency()
        
        # Find all tasks with no dependencies
        queue = deque()
        for name, degree in in_degree.items():
            if degree == 0:
                queue.append(name)
        
        order = []
        while queue:
            current = queue.popleft()
            order.append(current)
            
            for dependent in adj[current]:
                in_degree[dependent] -= 1
                if in_degree[dependent] == 0:
                    queue.append(dependent)
        
        if len(order) != len(self.tasks):
            # Cycle detected - find the tasks involved
            cycle_tasks = [name for name, deg in in_degree.items() if deg > 0]
            raise ValueError(
                f"Circular dependency detected among tasks: {cycle_tasks}"
            )
        
        return order
    
    def run(self, stop_on_failure: bool = True) -> dict[str, TaskStatus]:
        """
        Execute all tasks in dependency order.
        
        Args:
            stop_on_failure: If True, skip tasks dependent on failed tasks
            
        Returns:
            Dictionary mapping task names to their final status
        """
        order = self.get_execution_order()
        failed_tasks = set()
        
        for task_name in order:
            task = self.tasks[task_name]
            
            # Check if any dependency failed
            failed_deps = [d for d in task.dependencies if d in failed_tasks]
            if failed_deps and stop_on_failure:
                task.status = TaskStatus.SKIPPED
                failed_tasks.add(task_name)  # Propagate failure
                print(f"⏭️  Skipping '{task_name}' (depends on failed: {failed_deps})")
                continue
            
            # Execute the task
            print(f"▶️  Running '{task_name}'...")
            task.status = TaskStatus.RUNNING
            start_time = time.time()
            
            try:
                task.result = task.action()
                task.status = TaskStatus.COMPLETED
                task.duration = time.time() - start_time
                print(f"✅ Completed '{task_name}' in {task.duration:.2f}s")
            except Exception as e:
                task.status = TaskStatus.FAILED
                task.error = e
                task.duration = time.time() - start_time
                failed_tasks.add(task_name)
                print(f"❌ Failed '{task_name}': {e}")
        
        return {name: task.status for name, task in self.tasks.items()}
 
 
# Example usage
def main():
    scheduler = TaskScheduler()
    
    # Define tasks with dependencies
    scheduler.add_task("setup_db", lambda: print("  Creating database..."))
    scheduler.add_task("run_migrations", lambda: print("  Running migrations..."), 
                       dependencies=["setup_db"])
    scheduler.add_task("seed_data", lambda: print("  Seeding test data..."), 
                       dependencies=["run_migrations"])
    scheduler.add_task("start_server", lambda: print("  Starting server..."), 
                       dependencies=["run_migrations"])
    scheduler.add_task("run_tests", lambda: print("  Running test suite..."), 
                       dependencies=["seed_data", "start_server"])
    scheduler.add_task("generate_report", lambda: print("  Generating report..."), 
                       dependencies=["run_tests"])
    
    print("📋 Execution Plan:", scheduler.get_execution_order())
    print("
🚀 Starting execution:
")
    results = scheduler.run()
    print("
📊 Final Status:", results)
 
if __name__ == "__main__":
    main()

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

from collections import deque
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass
from typing import Callable
import time
 
@dataclass
class ParallelTask:
    name: str
    action: Callable
    dependencies: list[str]
 
class ParallelScheduler:
    """
    Execute tasks in parallel where dependencies allow.
    Uses thread pool for concurrent execution.
    """
    
    def __init__(self, max_workers: int = 4):
        self.tasks: dict[str, ParallelTask] = {}
        self.max_workers = max_workers
    
    def add_task(self, name: str, action: Callable, dependencies: list[str] = None):
        self.tasks[name] = ParallelTask(name, action, dependencies or [])
    
    def compute_levels(self) -> dict[int, list[str]]:
        """
        Group tasks into parallel execution levels.
        Level 0: tasks with no dependencies
        Level n: tasks whose dependencies are all at level < n
        """
        in_degree = {name: len(task.dependencies) for name, task in self.tasks.items()}
        adj = {name: [] for name in self.tasks}
        for task in self.tasks.values():
            for dep in task.dependencies:
                adj[dep].append(task.name)
        
        levels = {}
        current_level = 0
        queue = deque([name for name, deg in in_degree.items() if deg == 0])
        
        while queue:
            # All tasks in current queue can run in parallel
            level_tasks = list(queue)
            levels[current_level] = level_tasks
            
            # Find next level's tasks
            next_queue = deque()
            for task_name in queue:
                for dependent in adj[task_name]:
                    in_degree[dependent] -= 1
                    if in_degree[dependent] == 0:
                        next_queue.append(dependent)
            
            queue = next_queue
            current_level += 1
        
        return levels
    
    def run_parallel(self) -> dict[str, float]:
        """
        Execute tasks with maximum parallelization.
        Returns dict mapping task name to execution time.
        """
        levels = self.compute_levels()
        timings = {}
        
        print(f"📊 Execution plan: {len(levels)} parallel levels")
        for level, tasks in levels.items():
            print(f"   Level {level}: {tasks}")
        print()
        
        with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
            for level, task_names in sorted(levels.items()):
                print(f"🔄 Starting level {level}: {task_names}")
                start = time.time()
                
                # Submit all tasks at this level
                futures = {
                    executor.submit(self.tasks[name].action): name 
                    for name in task_names
                }
                
                # Wait for all to complete
                for future in as_completed(futures):
                    name = futures[future]
                    try:
                        future.result()
                        print(f"   ✅ {name} completed")
                    except Exception as e:
                        print(f"   ❌ {name} failed: {e}")
                
                level_time = time.time() - start
                for name in task_names:
                    timings[name] = level_time
                    
                print(f"   Level {level} completed in {level_time:.2f}s
")
        
        return timings
 
 
# Example: Parallel build with simulated work
def main():
    import random
    
    def make_work(name, duration):
        def work():
            time.sleep(duration)
            return f"{name} done"
        return work
    
    scheduler = ParallelScheduler(max_workers=4)
    
    # Multiple independent source tasks
    scheduler.add_task("parse_config", make_work("parse_config", 0.5))
    scheduler.add_task("load_cache", make_work("load_cache", 0.3))
    scheduler.add_task("check_deps", make_work("check_deps", 0.4))
    
    # Tasks depending on sources
    scheduler.add_task("compile_a", make_work("compile_a", 1.0), 
                       ["parse_config"])
    scheduler.add_task("compile_b", make_work("compile_b", 0.8), 
                       ["parse_config"])
    scheduler.add_task("compile_c", make_work("compile_c", 1.2), 
                       ["load_cache"])
    
    # Final linking depends on all compilations
    scheduler.add_task("link", make_work("link", 0.5), 
                       ["compile_a", "compile_b", "compile_c"])
    
    start = time.time()
    scheduler.run_parallel()
    total = time.time() - start
    print(f"⏱️  Total parallel time: {total:.2f}s")
    print(f"   Sequential would be: ~4.7s")
    print(f"   Speedup: {4.7/total:.1f}x")
 
if __name__ == "__main__":
    main()

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

from pathlib import Path
from dataclasses import dataclass
from collections import deque
import hashlib
import time
 
@dataclass
class SourceFile:
    path: str
    dependencies: list[str]
    content_hash: str = ""
    last_compiled_hash: str = ""
 
class BuildSystem:
    """
    A simple build system demonstrating topological sort + change detection.
    """
    
    def __init__(self):
        self.files: dict[str, SourceFile] = {}
        self.compiled_order: list[str] = []
    
    def add_file(self, path: str, dependencies: list[str], content: str):
        """Register a source file with its dependencies."""
        content_hash = hashlib.md5(content.encode()).hexdigest()[:8]
        self.files[path] = SourceFile(
            path=path, 
            dependencies=dependencies,
            content_hash=content_hash
        )
    
    def _needs_recompile(self, path: str) -> bool:
        """Check if file needs recompilation."""
        f = self.files[path]
        # File changed?
        if f.content_hash != f.last_compiled_hash:
            return True
        # Any dependency needs recompile?
        for dep in f.dependencies:
            if self._needs_recompile(dep):
                return True
        return False
    
    def get_build_order(self) -> list[str]:
        """Compute build order via topological sort."""
        in_degree = {path: len(f.dependencies) for path, f in self.files.items()}
        adj = {path: [] for path in self.files}
        for path, f in self.files.items():
            for dep in f.dependencies:
                adj[dep].append(path)
        
        queue = deque([p for p, deg in in_degree.items() if deg == 0])
        order = []
        
        while queue:
            current = queue.popleft()
            order.append(current)
            for dependent in adj[current]:
                in_degree[dependent] -= 1
                if in_degree[dependent] == 0:
                    queue.append(dependent)
        
        if len(order) != len(self.files):
            raise ValueError("Circular dependency in build graph!")
        
        return order
    
    def build(self, incremental: bool = True) -> list[str]:
        """
        Build all files in dependency order.
        
        Returns list of files that were compiled.
        """
        order = self.get_build_order()
        compiled = []
        
        for path in order:
            f = self.files[path]
            
            if incremental:
                # Check if any dependency was just compiled
                deps_compiled = any(d in compiled for d in f.dependencies)
                file_changed = f.content_hash != f.last_compiled_hash
                
                if not file_changed and not deps_compiled:
                    print(f"   ⏭️  {path} (unchanged)")
                    continue
            
            # Simulate compilation
            print(f"   🔨 Compiling {path}...")
            time.sleep(0.1)  # Simulated work
            
            f.last_compiled_hash = f.content_hash
            compiled.append(path)
        
        return compiled
    
    def update_file(self, path: str, new_content: str):
        """Update a file's content (simulates editing)."""
        if path in self.files:
            self.files[path].content_hash = hashlib.md5(new_content.encode()).hexdigest()[:8]
 
 
# Example build
def main():
    build = BuildSystem()
    
    # types.h - no dependencies
    build.add_file("types.h", [], "typedef int MyInt;")
    
    # utils.h depends on types.h
    build.add_file("utils.h", ["types.h"], "#include <types.h>")
    
    # logger.h - independent
    build.add_file("logger.h", [], "void log(char* msg);")
    
    # main.c depends on utils.h and logger.h
    build.add_file("main.c", ["utils.h", "logger.h"], "main() {}")
    
    # app.c depends on main.c output
    build.add_file("app.c", ["main.c"], "app entry")
    
    print("🏗️  Build Order:", build.get_build_order())
    print("
📦 Initial Build:")
    built = build.build(incremental=False)
    print(f"   Built {len(built)} files
")
    
    print("📦 Rebuild (no changes):")
    built = build.build(incremental=True)
    print(f"   Built {len(built)} files
")
    
    print("✏️  Modifying utils.h...")
    build.update_file("utils.h", "new utils content")
    
    print("📦 Incremental Rebuild:")
    built = build.build(incremental=True)
    print(f"   Built {len(built)} files (utils.h, main.c, app.c)")
 
if __name__ == "__main__":
    main()

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

from collections import deque
 
def can_finish(num_courses: int, prerequisites: list[list[int]]) -> bool:
    """
    LeetCode 207: Course Schedule
    Determine if it's possible to finish all courses.
    """
    # Build adjacency list and in-degree count
    adj = [[] for _ in range(num_courses)]
    in_degree = [0] * num_courses
    
    for course, prereq in prerequisites:
        adj[prereq].append(course)  # prereq → course
        in_degree[course] += 1
    
    # Kahn's algorithm
    queue = deque([i for i in range(num_courses) if in_degree[i] == 0])
    completed = 0
    
    while queue:
        current = queue.popleft()
        completed += 1
        
        for next_course in adj[current]:
            in_degree[next_course] -= 1
            if in_degree[next_course] == 0:
                queue.append(next_course)
    
    return completed == num_courses
 
 
def find_order(num_courses: int, prerequisites: list[list[int]]) -> list[int]:
    """
    LeetCode 210: Course Schedule II
    Return a valid order to take all courses, or empty list if impossible.
    """
    adj = [[] for _ in range(num_courses)]
    in_degree = [0] * num_courses
    
    for course, prereq in prerequisites:
        adj[prereq].append(course)
        in_degree[course] += 1
    
    queue = deque([i for i in range(num_courses) if in_degree[i] == 0])
    order = []
    
    while queue:
        current = queue.popleft()
        order.append(current)
        
        for next_course in adj[current]:
            in_degree[next_course] -= 1
            if in_degree[next_course] == 0:
                queue.append(next_course)
    
    return order if len(order) == num_courses else []
 
 
# Examples
print("Example 1: Can finish 2 courses with prereqs [[1,0]]?")
print(f"  Result: {can_finish(2, [[1,0]])}")  # True: take 0, then 1
 
print("
Example 2: Can finish with circular deps [[1,0],[0,1]]?")
print(f"  Result: {can_finish(2, [[1,0], [0,1]])}")  # False: cycle
 
print("
Example 3: Order for 4 courses [[1,0],[2,0],[3,1],[3,2]]?")
print(f"  Result: {find_order(4, [[1,0],[2,0],[3,1],[3,2]])}")  # [0,1,2,3] or [0,2,1,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

from collections import deque
 
def alien_order(words: list[str]) -> str:
    """
    Derive the character ordering from a sorted alien dictionary.
    Returns the order string, or empty string if invalid.
    """
    # Step 1: Build the graph
    # Vertices: all unique characters
    # Edges: derived from adjacent word comparisons
    
    # Initialize adjacency list for all characters
    adj = {c: set() for word in words for c in word}
    in_degree = {c: 0 for c in adj}
    
    # Compare adjacent words to derive ordering
    for i in range(len(words) - 1):
        word1, word2 = words[i], words[i + 1]
        
        # Edge case: prefix should come before longer word
        if len(word1) > len(word2) and word1.startswith(word2):
            return ""  # Invalid input
        
        # Find first differing character
        for c1, c2 in zip(word1, word2):
            if c1 != c2:
                # c1 comes before c2 in alien alphabet
                if c2 not in adj[c1]:
                    adj[c1].add(c2)
                    in_degree[c2] += 1
                break  # Only first difference matters
    
    # Step 2: Topological sort via Kahn's algorithm
    queue = deque([c for c in in_degree if in_degree[c] == 0])
    result = []
    
    while queue:
        current = queue.popleft()
        result.append(current)
        
        for neighbor in adj[current]:
            in_degree[neighbor] -= 1
            if in_degree[neighbor] == 0:
                queue.append(neighbor)
    
    # Check if all characters were processed
    if len(result) != len(adj):
        return ""  # Cycle detected - invalid input
    
    return ''.join(result)
 
 
# Examples
print("Example 1: ['wrt','wrf','er','ett','rftt']")
print(f"  Order: {alien_order(['wrt','wrf','er','ett','rftt'])}")  # "wertf"
 
print("
Example 2: ['z','x']")
print(f"  Order: {alien_order(['z','x'])}")  # "zx"
 
print("
Example 3: ['z','x','z'] (invalid - cycle)")
print(f"  Order: {alien_order(['z','x','z'])}")  # "" (z<x and x<z is impossible)