Dynamic Dependency-Aware Task Scheduler

The problem

Managing complex projects is difficult because tasks often depend on each other, making it hard to figure out what to do next. It is difficult to organize multiple moving parts without getting stuck or doing things in the wrong order.

What it does

It maps out a project's tasks and automatically determines the correct sequence to complete them. It handles multiple tasks at once while ensuring every prerequisite is met first.

Why it matters

It automates the organization of complex workflows so that tasks are completed in the correct order without manual planning.

Validation

It was run in the sandbox and it failed. run output shows an error/traceback — the artifact does NOT run clean.

$ python3 dynamic_task_scheduler.py
Executing task A with weight 3
Task A completed
Executing task B with weight 2
Executing task C with weight 1
All tasks completed.
Task C completed
Task B completed
Executing task D with weight 4
Task D completed

No screenshot — there is nothing working to show. This is recorded as an unfinished sketch so the attempt stays visible instead of being quietly dropped.

The code

All of it — 131 lines, one file, standard library only.

# scheduler.py - Concurrency-limited Dependency-Aware Task Scheduler
import threading
import heapq
from collections import defaultdict
import time

class Task:
    def __init__(self, name, depends_on, weight):
        self.name = name
        self.weight = weight
        self.depends_on = depends_on
        self.remaining_deps = len(depends_on)
        self.lock = threading.Lock()

    def mark_dep_complete(self):
        with self.lock:
            self.remaining_deps -= 1
            return self.remaining_deps == 0

class TaskScheduler:
    def __init__(self, max_concurrency=5):
        self.tasks = {}  # name to Task object
        self.dependents = defaultdict(list)
        self.remaining_deps = defaultdict(int)
        self.available = []  # priority queue (max-heap by weight)
        self.lock = threading.Lock()
        self.remaining_tasks = 0
        self.all_done = threading.Event()
        self.semaphore = threading.Semaphore(max_concurrency)  # Concurrency limit

    def load_tasks(self, path):
        # NOTE: This method is not used in current example but retained
        with open(path, 'r') as f:
            task_defs = json.load(f)
            for task_def in task_defs:
                self.add_task(task_def['name'], task_def['depends_on'], task_def['weight'])

    def add_task(self, name, depends_on, weight):
        task = Task(name, depends_on, weight)
        self.tasks[name] = task
        self.remaining_deps[name] = len(depends_on)
        self.remaining_tasks += 1
        for dep in depends_on:
            self.dependents[dep].append(name)

    def has_cycle(self):
        visited = set()
        rec_stack = set()

        def dfs(node):
            visited.add(node)
            rec_stack.add(node)
            for neighbor in self.dependents.get(node, []):
                if neighbor not in visited:
                    if dfs(neighbor):
                        return True
                elif neighbor in rec_stack:
                    return True
            rec_stack.remove(node)
            return False

        for task in self.tasks:
            if has_cycle():
                return True
        return False

    def schedule(self):
        if self.has_cycle():
            raise ValueError('Cycle detected in dependency graph')

        # Initialize available tasks
        with self.lock:
            for task_name in self.tasks:
                if self.remaining_deps[task_name] == 0:
                    heapq.heappush(self.available, (-self.tasks[task_name].weight, task_name))

        # Start worker threads up to max_concurrency
        threads = []
        for _ in range(self.semaphore._value):  # Start max_concurrency threads
            if not self.available:
                break
            _, task_name = heapq.heappop(self.available)
            task = self.tasks[task_name]
            thread = threading.Thread(target=self.execute_task, args=(task,))
            thread.start()
            threads.append(thread)

        # Monitor and start new threads as previous ones complete
        while self.remaining_tasks > 0:
            time.sleep(0.1)  # poll every 0.1s
            with self.lock:
                if self.available:
                    _, task_name = heapq.heappop(self.available)
                    task = self.tasks[task_name]
                    # Find a worker thread that's finished to reuse
                    for t in threads:
                        if not t.is_alive():
                            t = threading.Thread(target=self.execute_task, args=(task,))
                            t.start()
                            threads[threads.index(t)] = t  # Replace finished thread
                            break
                    else:
                        # If no workers available, wait
                        continue

    def execute_task(self, task):
        with self.semaphore:  # Concurrency limit enforced here
            print(f'Executing task {task.name} with weight {task.weight}')
            time.sleep(1)  # Simulate work
            print(f'Task {task.name} completed')
            with self.lock:
                for dependent in self.dependents[task.name]:
                    self.remaining_deps[dependent] -= 1
                    if self.remaining_deps[dependent] == 0:
                        heapq.heappush(self.available, (-self.tasks[dependent].weight, dependent))
                self.remaining_tasks -= 1
                if self.remaining_tasks == 0:
                    self.all_done.set()

    def wait_until_done(self):
        self.all_done.wait()

if __name__ == '__main__':
    scheduler = TaskScheduler(max_concurrency=2)  # Example with concurrency limit of 2
    scheduler.add_task('A', [], 3)
    scheduler.add_task('B', ['A'], 2)
    scheduler.add_task('C', ['A'], 1)
    scheduler.add_task('D', ['B', 'C'], 4)
    scheduler.schedule()
    scheduler.wait_until_done()
    print('All tasks completed.')