有向无环图任务的并行执行

有向无环图(Directed Acyclic Graph, DAG)任务的并行执行可以通过拓扑排序和并行执行的方式来实现。下面是一个示例代码，展示了如何使用多线程并行执行DAG任务。

import threading

class Task:
    def __init__(self, id):
        self.id = id
        self.dependencies = []
        self.completed = False
        self.lock = threading.Lock()

    def add_dependency(self, task):
        self.dependencies.append(task)

    def execute(self):
        # 执行任务的代码
        print("Task", self.id, "is executing")

        # 模拟任务执行时间
        import time
        time.sleep(1)

        # 标记任务已完成
        self.completed = True
        print("Task", self.id, "is completed")

    def is_ready(self):
        # 检查任务的所有依赖是否已完成
        for dependency in self.dependencies:
            if not dependency.completed:
                return False
        return True

def execute_task(task):
    # 等待任务的所有依赖都完成
    while not task.is_ready():
        pass

    # 获取任务的锁
    task.lock.acquire()

    # 执行任务
    task.execute()

    # 释放锁
    task.lock.release()

def parallel_execution(tasks):
    threads = []

    # 创建线程并执行任务
    for task in tasks:
        thread = threading.Thread(target=execute_task, args=(task,))
        thread.start()
        threads.append(thread)

    # 等待所有线程结束
    for thread in threads:
        thread.join()

# 创建任务
task1 = Task(1)
task2 = Task(2)
task3 = Task(3)
task4 = Task(4)

# 设置任务之间的依赖关系
task2.add_dependency(task1)
task3.add_dependency(task1)
task4.add_dependency(task2)
task4.add_dependency(task3)

# 将任务添加到列表中
tasks = [task1, task2, task3, task4]

# 并行执行任务
parallel_execution(tasks)

上述代码中，首先定义了一个Task类，表示一个任务。每个任务有一个唯一的id，一个列表dependencies用来存储该任务的依赖关系，一个completed标志表示任务是否已完成，以及一个lock用来实现线程安全。

然后定义了execute_task函数，用来执行一个任务。该函数会检查任务的所有依赖是否已完成，如果是，则获取任务的锁并执行任务，最后释放锁。

最后定义了parallel_execution函数，用来并行执行一组任务。该函数会创建多个线程，每个线程执行一个任务。在执行任务前，会先等待任务的所有依赖都完成。

在示例中，创建了4个任务，并设置它们之间的依赖关系。然后将任务添加到列表中，并调用parallel_execution函数并行执行这些任务。

输出结果可能会有所不同，但总体上会按照拓扑排序的顺序执行任务。